spin_factor_outbound_networking/
lib.rs

1mod allowed_hosts;
2pub mod runtime_config;
3mod tls;
4
5use std::{collections::HashMap, sync::Arc};
6
7use futures_util::FutureExt as _;
8use opentelemetry_semantic_conventions::attribute as otel_attribute;
9use spin_factor_variables::VariablesFactor;
10use spin_factor_wasi::{SocketAddrUse, SocketPermitState, WasiFactor};
11use spin_factors::{
12    ConfigureAppContext, Error, Factor, FactorInstanceBuilder, PrepareContext, RuntimeFactors,
13    anyhow::{self, Context},
14};
15use spin_locked_app::APP_NAME_KEY;
16use spin_outbound_networking_config::allowed_hosts::{DisallowedHostHandler, OutboundAllowedHosts};
17use url::Url;
18
19use crate::{
20    allowed_hosts::allowed_outbound_hosts, runtime_config::RuntimeConfig, tls::TlsClientConfigs,
21};
22pub use allowed_hosts::validate_service_chaining_for_components;
23pub use spin_connection_semaphore::{ConnectionPermit, ConnectionSemaphore, LimitedSemaphore};
24
25pub use crate::tls::{ComponentTlsClientConfigs, TlsClientConfig};
26use config::allowed_hosts::AllowedHostsConfig;
27use config::blocked_networks::BlockedNetworks;
28pub use spin_outbound_networking_config as config;
29
30#[derive(Default)]
31pub struct OutboundNetworkingFactor {
32    disallowed_host_handler: Option<Arc<dyn DisallowedHostHandler>>,
33}
34
35impl OutboundNetworkingFactor {
36    pub fn new() -> Self {
37        Self::default()
38    }
39
40    /// Sets a handler to be called when a request is disallowed by an
41    /// instance's configured `allowed_outbound_hosts`.
42    pub fn set_disallowed_host_handler(&mut self, handler: impl DisallowedHostHandler + 'static) {
43        self.disallowed_host_handler = Some(Arc::new(handler));
44    }
45}
46
47impl Factor for OutboundNetworkingFactor {
48    type RuntimeConfig = RuntimeConfig;
49    type AppState = AppState;
50    type InstanceBuilder = InstanceBuilder;
51
52    fn configure_app<T: RuntimeFactors>(
53        &self,
54        mut ctx: ConfigureAppContext<T, Self>,
55    ) -> anyhow::Result<Self::AppState> {
56        // Extract allowed_outbound_hosts for all components
57        let component_allowed_hosts = ctx
58            .app()
59            .components()
60            .map(|component| {
61                Ok((
62                    component.id().to_string(),
63                    allowed_outbound_hosts(&component)?
64                        .into_boxed_slice()
65                        .into(),
66                ))
67            })
68            .collect::<anyhow::Result<_>>()?;
69
70        let RuntimeConfig {
71            client_tls_configs,
72            blocked_ip_networks: block_networks,
73            block_private_networks,
74            max_socket_connections,
75            max_total_connections,
76            wait_timeout,
77        } = ctx.take_runtime_config().unwrap_or_default();
78
79        let blocked_networks = BlockedNetworks::new(block_networks, block_private_networks);
80        let tls_client_configs = TlsClientConfigs::new(client_tls_configs)?;
81        // Build the shared global semaphore from its limit, so the two are bound together from
82        // creation as the pair is plumbed into per-factor `ConnectionSemaphore`s.
83        let global_connection_semaphore = max_total_connections.map(LimitedSemaphore::new);
84
85        if let (Some(socket_cap), Some(global_cap)) =
86            (max_socket_connections, max_total_connections)
87            && socket_cap > global_cap
88        {
89            tracing::warn!(
90                "outbound_networking max_socket_connections ({socket_cap}) exceeds \
91                 max_total_connections ({global_cap}); the global limit will be the effective \
92                 cap for TCP/UDP sockets"
93            );
94        }
95
96        let app_id: Arc<str> = ctx
97            .app()
98            .get_metadata(APP_NAME_KEY)?
99            .unwrap_or_else(|| "<unnamed>".into())
100            .into();
101
102        let socket_connection_semaphore =
103            if max_socket_connections.is_some() || global_connection_semaphore.is_some() {
104                Some(ConnectionSemaphore::new(
105                    global_connection_semaphore.clone(),
106                    max_socket_connections.map(LimitedSemaphore::new),
107                    "wasi-sockets",
108                    app_id.clone(),
109                    wait_timeout,
110                ))
111            } else {
112                None
113            };
114
115        Ok(AppState {
116            component_allowed_hosts,
117            blocked_networks,
118            tls_client_configs,
119            socket_connection_semaphore,
120            global_connection_semaphore,
121            app_id,
122        })
123    }
124
125    fn prepare<T: RuntimeFactors>(
126        &self,
127        mut ctx: PrepareContext<T, Self>,
128    ) -> anyhow::Result<Self::InstanceBuilder> {
129        let hosts = ctx
130            .app_state()
131            .component_allowed_hosts
132            .get(ctx.app_component().id())
133            .cloned()
134            .context("missing component allowed hosts")?;
135        let resolver = ctx
136            .instance_builder::<VariablesFactor>()?
137            .expression_resolver()
138            .clone();
139        let component_ids = ctx
140            .app_component()
141            .app
142            .components()
143            .map(|c| c.id().to_string())
144            .collect::<Vec<_>>();
145        let allowed_hosts_future = async move {
146            let prepared = resolver.prepare().await.inspect_err(|err| {
147                tracing::error!(
148                    %err, "error.type" = "variable_resolution_failed",
149                    "Error resolving variables when checking request against allowed outbound hosts",
150                );
151            })?;
152            AllowedHostsConfig::parse(&hosts, &prepared, &component_ids).inspect_err(|err| {
153                tracing::error!(
154                    %err, "error.type" = "invalid_allowed_hosts",
155                    "Error parsing allowed outbound hosts",
156                );
157            })
158        }
159        .map(|res| res.map(Arc::new).map_err(Arc::new))
160        .boxed()
161        .shared();
162        let allowed_hosts = OutboundAllowedHosts::new(
163            allowed_hosts_future.clone(),
164            self.disallowed_host_handler.clone(),
165        );
166        let blocked_networks = ctx.app_state().blocked_networks.clone();
167        let permit_state = ctx
168            .app_state()
169            .socket_connection_semaphore
170            .clone()
171            .map(SocketPermitState::new);
172
173        match ctx.instance_builder::<WasiFactor>() {
174            Ok(wasi_builder) => {
175                if let Some(state) = permit_state {
176                    wasi_builder.set_socket_permit_state(state);
177                }
178
179                let allowed_hosts = allowed_hosts.clone();
180                wasi_builder.outbound_socket_addr_check(move |addr, addr_use| {
181                    let allowed_hosts = allowed_hosts.clone();
182                    let blocked_networks = blocked_networks.clone();
183                    async move {
184                        let scheme = match addr_use {
185                            SocketAddrUse::TcpBind => return false,
186                            SocketAddrUse::TcpConnect => "tcp",
187                            SocketAddrUse::UdpBind
188                            | SocketAddrUse::UdpConnect
189                            | SocketAddrUse::UdpOutgoingDatagram => "udp",
190                        };
191                        if !allowed_hosts
192                            .check_url(&addr.to_string(), scheme)
193                            .await
194                            .unwrap_or(
195                                // TODO: should this trap (somehow)?
196                                false,
197                            )
198                        {
199                            return false;
200                        }
201                        if blocked_networks.is_blocked(&addr) {
202                            tracing::error!(
203                                "error.type" = "destination_ip_prohibited",
204                                ?addr,
205                                "destination IP prohibited by runtime config"
206                            );
207                            return false;
208                        }
209                        true
210                    }
211                });
212            }
213            Err(Error::NoSuchFactor(_)) => (), // no WasiFactor to configure; that's OK
214            Err(err) => return Err(err.into()),
215        }
216
217        let component_tls_configs = ctx
218            .app_state()
219            .tls_client_configs
220            .get_component_tls_configs(ctx.app_component().id());
221
222        Ok(InstanceBuilder {
223            allowed_hosts,
224            blocked_networks: ctx.app_state().blocked_networks.clone(),
225            component_tls_client_configs: component_tls_configs,
226        })
227    }
228}
229
230pub struct AppState {
231    /// Component ID -> Allowed host list
232    component_allowed_hosts: HashMap<String, Arc<[String]>>,
233    /// Blocked IP networks
234    blocked_networks: BlockedNetworks,
235    /// TLS client configs
236    tls_client_configs: TlsClientConfigs,
237    /// Pre-built semaphore for TCP/UDP socket quota enforcement (global + socket-specific).
238    /// `None` means no limits are configured.
239    socket_connection_semaphore: Option<ConnectionSemaphore>,
240    /// App-wide semaphore capping total concurrent outbound connections across ALL types,
241    /// paired with its configured limit (the latter is used for warning comparisons in other
242    /// factors). `None` means unlimited.
243    global_connection_semaphore: Option<LimitedSemaphore>,
244    /// Identifier of this app, used for tenant attribution on tracing events emitted by
245    /// outbound factors' connection semaphores. Resolved once at configure-app time.
246    app_id: Arc<str>,
247}
248
249impl AppState {
250    /// Returns the app identifier, used by outbound factors when building per-factor
251    /// connection semaphores so rejection tracing events carry tenant attribution.
252    pub fn app_id(&self) -> Arc<str> {
253        self.app_id.clone()
254    }
255}
256
257/// Builds a [`ConnectionSemaphore`] for an outbound factor, incorporating the optional global
258/// connection limit from the networking factor's app state.
259///
260/// Emits a warning when the per-factor limit exceeds the global cap (the global limit would
261/// be the effective ceiling in that case).
262///
263/// The app identifier is taken from the networking app state (or `<unnamed>` when networking is
264/// absent) and plumbed through to the semaphore so that rejection tracing events can carry tenant
265/// attribution without app identity appearing on metric labels.
266pub fn build_connection_semaphore(
267    networking: Option<&AppState>,
268    factor: &'static str,
269    factor_limit: Option<usize>,
270    wait_timeout: Option<std::time::Duration>,
271) -> ConnectionSemaphore {
272    let app_id = networking
273        .map(|n| n.app_id())
274        .unwrap_or_else(|| Arc::from("<unnamed>"));
275    let global = networking.and_then(|n| n.global_connection_semaphore.clone());
276    if let (Some(per_factor), Some(global_limit)) =
277        (factor_limit, global.as_ref().map(LimitedSemaphore::limit))
278        && per_factor > global_limit
279    {
280        tracing::warn!(
281            "outbound_{factor} max_connections ({per_factor}) exceeds global \
282             max_total_connections ({global_limit}); the global limit will be the \
283             effective cap"
284        );
285    }
286    ConnectionSemaphore::new(
287        global,
288        factor_limit.map(LimitedSemaphore::new),
289        factor,
290        app_id,
291        wait_timeout,
292    )
293}
294
295pub struct InstanceBuilder {
296    allowed_hosts: OutboundAllowedHosts,
297    blocked_networks: BlockedNetworks,
298    component_tls_client_configs: ComponentTlsClientConfigs,
299}
300
301impl InstanceBuilder {
302    pub fn allowed_hosts(&self) -> OutboundAllowedHosts {
303        self.allowed_hosts.clone()
304    }
305
306    pub fn blocked_networks(&self) -> BlockedNetworks {
307        self.blocked_networks.clone()
308    }
309
310    pub fn component_tls_configs(&self) -> ComponentTlsClientConfigs {
311        self.component_tls_client_configs.clone()
312    }
313}
314
315impl FactorInstanceBuilder for InstanceBuilder {
316    type InstanceState = ();
317
318    fn build(self) -> anyhow::Result<Self::InstanceState> {
319        Ok(())
320    }
321}
322
323/// Records the address host, port, and database as fields on the current tracing span.
324///
325/// This should only be called from within a function that has been instrumented with a span.
326///
327/// The following fields must be pre-declared as empty on the span or they will not show up.
328/// ```
329/// use tracing::field::Empty;
330/// use opentelemetry_semantic_conventions::attribute as otel_attribute;
331/// #[tracing::instrument(fields({otel_attribute::SERVER_ADDRESS} = Empty, {otel_attribute::SERVER_PORT} = Empty, {otel_attribute::DB_NAMESPACE} = Empty))]
332/// fn open() {}
333/// ```
334pub fn record_address_fields(address: &str) {
335    if let Ok(url) = Url::parse(address) {
336        let span = tracing::Span::current();
337        span.record(
338            otel_attribute::SERVER_ADDRESS,
339            url.host_str().unwrap_or_default(),
340        );
341        span.record(otel_attribute::SERVER_PORT, url.port().unwrap_or_default());
342        span.record(
343            otel_attribute::DB_NAMESPACE,
344            url.path().trim_start_matches('/'),
345        );
346    }
347}
spin_factor_outbound_networking/lib.rs

spin_factor_outbound_networking/
lib.rs
