thread through --allow-local flag to enable/disable local connectors

We'd like them to be available in local contexts, like `flowctl` and
also local stacks, while being disabled in production deployments.

Author: jgraettinger

crates/agent/src/main.rs

@@ -43,6 +43,9 @@ struct Args {
     /// Email address of user which provisions and maintains tenant accounts.
     #[clap(long = "accounts-email", default_value = "[email protected]")]
     accounts_email: String,
+    /// Allow local connectors. True for local stacks, and false otherwise.
+    #[clap(long = "allow-local", default_value = "false")]
+    allow_local: bool,
 }
 
 #[tokio::main]
@@ -96,6 +99,7 @@ async fn main() -> Result<(), anyhow::Error> {
         vec![
             Box::new(agent::PublishHandler::new(
                 &args.accounts_email,
+                args.allow_local,
                 &bindir,
                 &args.broker_address,
                 &builds_root,

crates/agent/src/publications.rs

@@ -60,6 +60,7 @@ impl JobStatus {
 /// A PublishHandler is a Handler which publishes catalog specifications.
 pub struct PublishHandler {
     agent_user_email: String,
+    allow_local: bool,
     bindir: String,
     broker_address: url::Url,
     builds_root: url::Url,
@@ -72,6 +73,7 @@ pub struct PublishHandler {
 impl PublishHandler {
     pub fn new(
         agent_user_email: impl Into<String>,
+        allow_local: bool,
         bindir: &str,
         broker_address: &url::Url,
         builds_root: &url::Url,
@@ -82,6 +84,7 @@ impl PublishHandler {
     ) -> Self {
         Self {
             agent_user_email: agent_user_email.into(),
+            allow_local,
             bindir: bindir.to_string(),
             broker_address: broker_address.clone(),
             builds_root: builds_root.clone(),
@@ -321,6 +324,7 @@ impl PublishHandler {
         let tmpdir = tmpdir_handle.path();
 
         let build_output = builds::build_catalog(
+            self.allow_local,
             &self.builds_root,
             &draft_catalog,
             &self.connector_network,

crates/agent/src/publications/builds.rs

@@ -117,6 +117,7 @@ impl BuildOutput {
 }
 
 pub async fn build_catalog(
+    allow_local: bool,
     builds_root: &url::Url,
     catalog: &models::Catalog,
     connector_network: &str,
@@ -146,6 +147,7 @@ pub async fn build_catalog(
     let source = url::Url::parse("file:///flow.json").unwrap();
 
     let managed_build = build::managed_build(
+        allow_local,
         build_id.clone(),
         connector_network.to_string(),
         Box::new(control_plane),

crates/agent/src/publications/specs.rs

@@ -754,6 +754,7 @@ mod test {
 
         let mut handler = PublishHandler::new(
             "[email protected]",
+            false,
             "",
             &bs_url,
             &bs_url,

crates/build/src/lib.rs

@@ -94,6 +94,7 @@ pub async fn load(source: &url::Url, file_root: &Path) -> tables::Sources {
 /// * If `generate_ops_collections` is set, then ops collections are added into `sources`.
 /// * If any of `noop_*` is true, then validations are skipped for connectors of that type.
 pub async fn validate<L>(
+    allow_local: bool,
     build_id: &str,
     connector_network: &str,
     control_plane: &dyn validation::ControlPlane,
@@ -116,6 +117,7 @@ where
     ::sources::inline_sources(&mut sources);
 
     let runtime = runtime::Runtime::new(
+        allow_local,
         connector_network.to_string(),
         log_handler,
         None,
@@ -167,6 +169,7 @@ where
 /// components but not the `flowctl` CLI, which requires finer-grain
 /// control over build behavior.
 pub async fn managed_build<L>(
+    allow_local: bool,
     build_id: String,
     connector_network: String,
     control_plane: Box<dyn validation::ControlPlane>,
@@ -179,6 +182,7 @@ where
     L: Fn(&ops::Log) + Send + Sync + Clone + 'static,
 {
     let (sources, validations) = validate(
+        allow_local,
         &build_id,
         &connector_network,
         &*control_plane,

crates/flowctl/src/local_specs.rs

@@ -77,6 +77,7 @@ async fn validate(
     let project_root = build::project_root(source);
 
     let (sources, mut validations) = build::validate(
+        true, // Allow local connectors.
         "local-build",
         "", // Use default connector network.
         &Resolver { client },

crates/flowctl/src/preview/mod.rs

@@ -150,6 +150,7 @@ impl Preview {
             .await?;
 
         let mut responses_rx = runtime::Runtime::new(
+            true, // Allow local connectors.
             String::new(),
             ops::tracing_log_handler,
             None,

crates/flowctl/src/raw/mod.rs

@@ -206,6 +206,7 @@ async fn do_build(ctx: &mut crate::CliContext, build: &Build) -> anyhow::Result<
     let project_root = build::project_root(&source_url);
 
     let build_result = build::managed_build(
+        true, // Allow local connectors.
         build_id.clone(),
         connector_network.clone(),
         Box::new(local_specs::Resolver { client }),

crates/proto-flow/src/runtime.rs

@@ -9,6 +9,8 @@ pub struct TaskServiceConfig {
     pub uds_path: ::prost::alloc::string::String,
     #[prost(string, tag = "4")]
     pub container_network: ::prost::alloc::string::String,
+    #[prost(bool, tag = "5")]
+    pub allow_local: bool,
 }
 /// ShuffleRequest is the request message of a Shuffle RPC.
 /// It's a description of a document shuffle,

crates/runtime/src/capture/mod.rs

@@ -88,6 +88,11 @@ where
                 &self.task_name,
             )
             .boxed(),
+            models::CaptureEndpoint::Local(_) if !self.allow_local => {
+                return Err(tonic::Status::failed_precondition(
+                    "Local connectors are not permitted in this context",
+                ))
+            }
             models::CaptureEndpoint::Local(_) => todo!(),
         };