AWS SigV4 signing for Apollo Connectors

apollo-router/src/configuration/connector.rs

@@ -0,0 +1,16 @@
+use std::collections::HashMap;
+
+use schemars::JsonSchema;
+use serde::Deserialize;
+use serde::Serialize;
+
+#[derive(Clone, Debug, Default, Deserialize, Serialize, JsonSchema)]
+#[serde(bound(deserialize = "T: Deserialize<'de>"))] // T does not need to be Default
+pub(crate) struct ConnectorConfiguration<T>
+where
+    T: Serialize + JsonSchema,
+{
+    // Map of subgraph_name.connector_source_name to configuration
+    #[serde(default)]
+    pub(crate) sources: HashMap<String, T>,
+}

apollo-router/src/configuration/mod.rs

@@ -55,6 +55,7 @@ use crate::plugins::subscription::APOLLO_SUBSCRIPTION_PLUGIN_NAME;
 use crate::uplink::UplinkConfig;
 use crate::ApolloRouterError;
 
+pub(crate) mod connector;
 pub(crate) mod cors;
 pub(crate) mod expansion;
 mod experimental;

apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__schema_generation.snap

@@ -1326,6 +1326,11 @@ expression: "&schema"
       "additionalProperties": false,
       "description": "Authentication",
       "properties": {
+        "connector": {
+          "$ref": "#/definitions/ConnectorConfiguration_for_AuthConfig",
+          "description": "#/definitions/ConnectorConfiguration_for_AuthConfig",
+          "nullable": true
+        },
         "router": {
           "$ref": "#/definitions/RouterConf",
           "description": "#/definitions/RouterConf",
@@ -1763,6 +1768,7 @@ expression: "&schema"
             "$ref": "#/definitions/AuthConfig",
             "description": "#/definitions/AuthConfig"
           },
+          "default": {},
           "description": "Create a configuration that will apply only to a specific subgraph.",
           "type": "object"
         }
@@ -1943,6 +1949,19 @@ expression: "&schema"
       },
       "type": "object"
     },
+    "ConnectorConfiguration_for_AuthConfig": {
+      "properties": {
+        "sources": {
+          "additionalProperties": {
+            "$ref": "#/definitions/AuthConfig",
+            "description": "#/definitions/AuthConfig"
+          },
+          "default": {},
+          "type": "object"
+        }
+      },
+      "type": "object"
+    },
     "ConnectorEventsConfig": {
       "additionalProperties": false,
       "properties": {

apollo-router/src/plugins/authentication/connector.rs

@@ -0,0 +1,50 @@
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use tower::ServiceBuilder;
+use tower::ServiceExt;
+
+use crate::plugins::authentication::subgraph::SigningParamsConfig;
+use crate::services::connector_service::ConnectorInfo;
+use crate::services::connector_service::ConnectorSourceRef;
+use crate::services::connector_service::CONNECTOR_INFO_CONTEXT_KEY;
+use crate::services::http::HttpRequest;
+
+pub(super) struct ConnectorAuth {
+    pub(super) signing_params: Arc<HashMap<ConnectorSourceRef, Arc<SigningParamsConfig>>>,
+}
+
+impl ConnectorAuth {
+    pub(super) fn http_client_service(
+        &self,
+        subgraph_name: &str,
+        service: crate::services::http::BoxService,
+    ) -> crate::services::http::BoxService {
+        let signing_params = self.signing_params.clone();
+        let subgraph_name = subgraph_name.to_string();
+        ServiceBuilder::new()
+            .map_request(move |req: HttpRequest| {
+                if let Ok(Some(connector_info)) = req
+                    .context
+                    .get::<&str, ConnectorInfo>(CONNECTOR_INFO_CONTEXT_KEY)
+                {
+                    if let Some(source_name) = connector_info.source_name {
+                        if let Some(signing_params) = signing_params
+                            .get(&ConnectorSourceRef::new(
+                                subgraph_name.clone(),
+                                source_name.clone(),
+                            ))
+                            .cloned()
+                        {
+                            req.context
+                                .extensions()
+                                .with_lock(|mut lock| lock.insert(signing_params));
+                        }
+                    }
+                }
+                req
+            })
+            .service(service)
+            .boxed()
+    }
+}

apollo-router/src/plugins/authentication/mod.rs

@@ -42,19 +42,24 @@ use self::jwks::JwksManager;
 use self::subgraph::SigningParams;
 use self::subgraph::SigningParamsConfig;
 use self::subgraph::SubgraphAuth;
+use crate::configuration::connector::ConnectorConfiguration;
 use crate::graphql;
 use crate::layers::ServiceBuilderExt;
 use crate::plugin::serde::deserialize_header_name;
 use crate::plugin::serde::deserialize_header_value;
-use crate::plugin::Plugin;
 use crate::plugin::PluginInit;
+use crate::plugin::PluginPrivate;
+use crate::plugins::authentication::connector::ConnectorAuth;
 use crate::plugins::authentication::jwks::JwkSetInfo;
 use crate::plugins::authentication::jwks::JwksConfig;
-use crate::register_plugin;
+use crate::plugins::authentication::subgraph::make_signing_params;
+use crate::plugins::authentication::subgraph::AuthConfig;
+use crate::services::connector_service::ConnectorSourceRef;
 use crate::services::router;
 use crate::services::APPLICATION_JSON_HEADER_VALUE;
 use crate::Context;
 
+mod connector;
 mod jwks;
 pub(crate) mod subgraph;
 
@@ -123,6 +128,7 @@ struct Router {
 struct AuthenticationPlugin {
     router: Option<Router>,
     subgraph: Option<SubgraphAuth>,
+    connector: Option<ConnectorAuth>,
 }
 
 #[derive(Clone, Debug, Deserialize, JsonSchema, serde_derive_default::Default)]
@@ -207,6 +213,8 @@ struct Conf {
     router: Option<RouterConf>,
     /// Subgraph configuration
     subgraph: Option<subgraph::Config>,
+    /// Connector configuration
+    connector: Option<ConnectorConfiguration<AuthConfig>>,
 }
 
 // We may support additional authentication mechanisms in future, so all
@@ -409,7 +417,7 @@ fn search_jwks(
 }
 
 #[async_trait::async_trait]
-impl Plugin for AuthenticationPlugin {
+impl PluginPrivate for AuthenticationPlugin {
     type Config = Conf;
 
     async fn new(init: PluginInit<Self::Config>) -> Result<Self, BoxError> {
@@ -491,7 +499,30 @@ impl Plugin for AuthenticationPlugin {
             None
         };
 
-        Ok(Self { router, subgraph })
+        let connector = if let Some(config) = init.config.connector {
+            let mut signing_params: HashMap<ConnectorSourceRef, Arc<SigningParamsConfig>> =
+                Default::default();
+            for (s, source_config) in config.sources {
+                let source_ref: ConnectorSourceRef = s.parse()?;
+                signing_params.insert(
+                    source_ref.clone(),
+                    make_signing_params(&source_config, &source_ref.subgraph_name)
+                        .await
+                        .map(Arc::new)?,
+                );
+            }
+            Some(ConnectorAuth {
+                signing_params: Arc::new(signing_params),
+            })
+        } else {
+            None
+        };
+
+        Ok(Self {
+            router,
+            subgraph,
+            connector,
+        })
     }
 
     fn router_service(&self, service: router::BoxService) -> router::BoxService {
@@ -532,6 +563,18 @@ impl Plugin for AuthenticationPlugin {
             service
         }
     }
+
+    fn http_client_service(
+        &self,
+        subgraph_name: &str,
+        service: crate::services::http::BoxService,
+    ) -> crate::services::http::BoxService {
+        if let Some(auth) = &self.connector {
+            auth.http_client_service(subgraph_name, service)
+        } else {
+            service
+        }
+    }
 }
 
 fn authenticate(
@@ -934,4 +977,4 @@ pub(crate) fn convert_algorithm(algorithm: Algorithm) -> KeyAlgorithm {
 //
 // In order to keep the plugin names consistent,
 // we use using the `Reverse domain name notation`
-register_plugin!("apollo", "authentication", AuthenticationPlugin);
+register_private_plugin!("apollo", "authentication", AuthenticationPlugin);

apollo-router/src/plugins/authentication/subgraph.rs

@@ -19,6 +19,7 @@ use http::HeaderMap;
 use http::Request;
 use schemars::JsonSchema;
 use serde::Deserialize;
+use serde::Serialize;
 use tokio::sync::mpsc::Sender;
 use tokio::task::JoinHandle;
 use tower::BoxError;
@@ -31,7 +32,7 @@ use crate::services::SubgraphRequest;
 
 /// Hardcoded Config using access_key and secret.
 /// Prefer using DefaultChain instead.
-#[derive(Clone, JsonSchema, Deserialize, Debug)]
+#[derive(Clone, JsonSchema, Deserialize, Serialize, Debug)]
 #[serde(rename_all = "snake_case", deny_unknown_fields)]
 pub(crate) struct AWSSigV4HardcodedConfig {
     /// The ID for this access key.
@@ -64,7 +65,7 @@ impl ProvideCredentials for AWSSigV4HardcodedConfig {
 }
 
 /// Configuration of the DefaultChainProvider
-#[derive(Clone, JsonSchema, Deserialize, Debug)]
+#[derive(Clone, JsonSchema, Deserialize, Serialize, Debug)]
 #[serde(deny_unknown_fields)]
 pub(crate) struct DefaultChainConfig {
     /// The AWS region this chain applies to.
@@ -78,7 +79,7 @@ pub(crate) struct DefaultChainConfig {
 }
 
 /// Specify assumed role configuration.
-#[derive(Clone, JsonSchema, Deserialize, Debug)]
+#[derive(Clone, JsonSchema, Deserialize, Serialize, Debug)]
 #[serde(deny_unknown_fields)]
 pub(crate) struct AssumeRoleProvider {
     /// Amazon Resource Name (ARN)
@@ -91,7 +92,7 @@ pub(crate) struct AssumeRoleProvider {
 }
 
 /// Configure AWS sigv4 auth.
-#[derive(Clone, JsonSchema, Deserialize, Debug)]
+#[derive(Clone, JsonSchema, Deserialize, Serialize, Debug)]
 #[serde(rename_all = "snake_case")]
 pub(crate) enum AWSSigV4Config {
     Hardcoded(AWSSigV4HardcodedConfig),
@@ -170,7 +171,7 @@ impl AWSSigV4Config {
     }
 }
 
-#[derive(Clone, Debug, JsonSchema, Deserialize)]
+#[derive(Clone, Debug, JsonSchema, Deserialize, Serialize)]
 #[serde(deny_unknown_fields)]
 pub(crate) enum AuthConfig {
     #[serde(rename = "aws_sig_v4")]

apollo-router/src/plugins/connectors/http_json_transport.rs

@@ -96,6 +96,11 @@ pub(crate) fn make_request(
 
     let (json_body, form_body, body, apply_to_errors) = if let Some(ref selection) = transport.body
     {
+        // The URL and headers use the $context above, but JSON Selection errors if it is present
+        let inputs = inputs
+            .into_iter()
+            .filter(|(k, _)| *k != "$context")
+            .collect();
         let (json_body, apply_to_errors) = selection.apply_with_vars(&json!({}), &inputs);
         let mut form_body = None;
         let body = if let Some(json_body) = json_body.as_ref() {

apollo-router/src/services/connector_service.rs

@@ -1,6 +1,7 @@
 //! Tower service for connectors.
 
 use std::collections::HashMap;
+use std::str::FromStr;
 use std::sync::Arc;
 use std::task::Poll;
 
@@ -83,6 +84,39 @@ impl From<&Connector> for ConnectorInfo {
     }
 }
 
+/// A reference to a unique Connector source.
+#[derive(Hash, Eq, PartialEq, Clone)]
+pub(crate) struct ConnectorSourceRef {
+    pub(crate) subgraph_name: String,
+    pub(crate) source_name: String,
+}
+
+impl ConnectorSourceRef {
+    pub(crate) fn new(subgraph_name: String, source_name: String) -> Self {
+        Self {
+            subgraph_name,
+            source_name,
+        }
+    }
+}
+
+impl FromStr for ConnectorSourceRef {
+    type Err = String;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        let mut parts = s.split('.');
+        let subgraph_name = parts
+            .next()
+            .ok_or(format!("Invalid connector source reference '{}'", s))?
+            .to_string();
+        let source_name = parts
+            .next()
+            .ok_or(format!("Invalid connector source reference '{}'", s))?
+            .to_string();
+        Ok(Self::new(subgraph_name, source_name))
+    }
+}
+
 impl tower::Service<ConnectRequest> for ConnectorService {
     type Response = ConnectResponse;
     type Error = BoxError;
@@ -184,7 +218,7 @@ async fn execute(
                 .insert(CONNECTOR_INFO_CONTEXT_KEY, ConnectorInfo::from(connector))
                 .is_err()
             {
-                error!("Failed to store connector info in context - instruments may be inaccurate");
+                error!("Failed to store connector info in context");
             }
             let original_subgraph_name = original_subgraph_name.clone();
             let request_limit = request_limit.clone();