feat(privacy): S7 — local-only enforcement across integrations + tools (#4441)

src/openhuman/composio/client.rs

@@ -194,9 +194,11 @@ impl ComposioClient {
         // Egress spine (privacy epic S2, #4436): a Composio tool call ships the
         // (already-normalized) arguments to the third-party provider — disclose
         // the transfer before the round-trip. S4 will add an approval arm here.
-        crate::openhuman::security::egress::emit_external_transfer(
-            crate::openhuman::security::egress::EgressDescriptor::composio(tool),
-        );
+        let egress = crate::openhuman::security::egress::EgressDescriptor::composio(tool);
+        // Local-only enforcement (privacy epic S7, #4441): refuse the external
+        // tool call under LocalOnly BEFORE disclosing or sending it.
+        crate::openhuman::security::egress::enforce_egress(&egress)?;
+        crate::openhuman::security::egress::emit_external_transfer(egress);
         // PR #1827 routes all execute-side argument normalization
         // (including the bare-date → RFC 3339 fix #1802 brought to
         // `normalize_calendar_query_args` on `main`) through the
@@ -237,9 +239,11 @@ impl ComposioClient {
         // Egress spine (privacy epic S2, #4436): see `execute_tool`. This is the
         // caller-owns-retry entry point (e.g. `auth_retry`), disjoint from
         // `execute_tool`, so each logical tool call emits exactly once.
-        crate::openhuman::security::egress::emit_external_transfer(
-            crate::openhuman::security::egress::EgressDescriptor::composio(tool),
-        );
+        let egress = crate::openhuman::security::egress::EgressDescriptor::composio(tool);
+        // Local-only enforcement (privacy epic S7, #4441): same gate as
+        // `execute_tool` — this disjoint entry point must block too.
+        crate::openhuman::security::egress::enforce_egress(&egress)?;
+        crate::openhuman::security::egress::emit_external_transfer(egress);
         let arguments = super::execute_prepare::prepare_execute_arguments(tool, arguments)
             .map_err(anyhow::Error::msg)?;
         tracing::debug!(tool = %tool, "[composio] execute_tool_once (no built-in retry)");

src/openhuman/composio/client_tests.rs

@@ -48,6 +48,45 @@ async fn authorize_rejects_empty_toolkit() {
     );
 }
 
+/// Privacy epic S7 (#4441): under LocalOnly, both execute entry points refuse
+/// the external tool call before any HTTP round-trip. The inner client points at
+/// an unreachable address, so a passing test proves the gate short-circuits
+/// before the network (a leaked call would fail with a transport error, not the
+/// local-only message).
+#[tokio::test]
+async fn execute_tool_blocked_under_local_only() {
+    // Thread-scoped LocalOnly (see `live_policy::test_privacy_scope`) — never
+    // mutates the process-global policy, so it can't race sibling mock tests.
+    let _mode = crate::openhuman::security::live_policy::test_privacy_scope(
+        crate::openhuman::config::PrivacyMode::LocalOnly,
+    );
+    let inner = Arc::new(crate::openhuman::integrations::IntegrationClient::new(
+        "http://127.0.0.1:0".into(),
+        "test".into(),
+    ));
+    let client = ComposioClient::new(inner);
+
+    let err = client
+        .execute_tool("GITHUB_CREATE_ISSUE", None)
+        .await
+        .unwrap_err();
+    assert!(
+        err.to_string()
+            .contains("Local-only privacy mode is active"),
+        "execute_tool: unexpected error: {err}"
+    );
+    let err_once = client
+        .execute_tool_once("GITHUB_CREATE_ISSUE", None)
+        .await
+        .unwrap_err();
+    assert!(
+        err_once
+            .to_string()
+            .contains("Local-only privacy mode is active"),
+        "execute_tool_once: unexpected error: {err_once}"
+    );
+}
+
 /// `authorize()` must reject a non-object `extra_params` before making any HTTP call.
 #[tokio::test]
 async fn authorize_rejects_non_object_extra_params() {

src/openhuman/composio/execute_dispatch.rs

@@ -156,6 +156,22 @@ pub async fn execute_composio_action_kind_with_connection(
         return Err("composio: tool slug must not be empty".to_string());
     }
 
+    // Local-only enforcement (privacy epic S7, #4441): gate ABOVE the
+    // backend-vs-direct branch. The per-client gate lives inside
+    // `ComposioClient::execute_tool{,_once}` and therefore only covers the
+    // Backend arm — the Direct arm (`direct_execute`) never reaches it, so
+    // without this a Composio tool call would still POST its arguments to
+    // Composio under LocalOnly in direct mode (the dual-path drift this very
+    // review flagged). A Composio execute ships the tool arguments off-device
+    // in BOTH modes, so refuse here — once, before either dispatch — for a
+    // single chokepoint. The Backend arm stays gated too (defense in depth).
+    if let Err(e) = crate::openhuman::security::egress::enforce_egress(
+        &crate::openhuman::security::egress::EgressDescriptor::composio(tool_trim),
+    ) {
+        tracing::debug!(tool = %tool_trim, "[composio][dispatch] local-only egress block");
+        return Err(e.to_string());
+    }
+
     let prepared = match prepare_execute_arguments(tool_trim, arguments) {
         Ok(args) => args,
         Err(msg) => {

src/openhuman/embeddings/cloud_adapter.rs

@@ -85,12 +85,44 @@ impl EmbeddingProvider for OpenHumanCloudEmbedding {
     async fn embed(&self, texts: &[&str]) -> anyhow::Result<Vec<Vec<f32>>> {
         // Egress spine (privacy epic S2, #4436): the input texts leave the
         // device for the cloud embedding backend — disclose before the request.
-        crate::openhuman::security::egress::emit_external_transfer(
-            crate::openhuman::security::egress::EgressDescriptor::embedding(
-                "cloud",
-                self.inner.model_id(),
-            ),
+        let egress = crate::openhuman::security::egress::EgressDescriptor::embedding(
+            "cloud",
+            self.inner.model_id(),
         );
+        // Local-only enforcement (privacy epic S7, #4441): refuse cloud
+        // embedding under LocalOnly before disclosing or sending the texts.
+        crate::openhuman::security::egress::enforce_egress(&egress)?;
+        crate::openhuman::security::egress::emit_external_transfer(egress);
         self.inner.embed(texts).await
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Privacy epic S7 (#4441): under LocalOnly, `embed` refuses before touching
+    /// the inner cloud transport (so no network / no auth is needed to observe
+    /// the block). Uses the thread-scoped privacy override so it never mutates
+    /// the process-global policy that sibling tests read.
+    #[tokio::test]
+    async fn embed_blocked_under_local_only() {
+        let _mode = crate::openhuman::security::live_policy::test_privacy_scope(
+            crate::openhuman::config::PrivacyMode::LocalOnly,
+        );
+        let provider = OpenHumanCloudEmbedding::new(
+            Some("http://127.0.0.1:0".into()),
+            Some(std::env::temp_dir().join("openhuman_embeddings_localonly_state")),
+            false,
+            DEFAULT_CLOUD_EMBEDDING_MODEL,
+            DEFAULT_CLOUD_EMBEDDING_DIMENSIONS,
+        );
+
+        let err = provider.embed(&["hello"]).await.unwrap_err();
+        assert!(
+            err.to_string()
+                .contains("Local-only privacy mode is active"),
+            "unexpected error: {err}"
+        );
+    }
+}

src/openhuman/integrations/client.rs

@@ -240,15 +240,30 @@ fn parse_content_disposition_filename(value: &str) -> Option<String> {
     None
 }
 
+/// Build the egress descriptor for a managed-backend round-trip. The query
+/// string is stripped so only the endpoint (the "to where") is described, never
+/// any data carried in the query.
+fn backend_egress_descriptor(path: &str) -> crate::openhuman::security::egress::EgressDescriptor {
+    let endpoint = path.split('?').next().unwrap_or(path);
+    crate::openhuman::security::egress::EgressDescriptor::integration(endpoint)
+}
+
 /// Egress spine (privacy epic S2, #4436): disclose an OpenHuman managed-backend
-/// round-trip before it leaves the device. The query string is stripped so only
-/// the endpoint (the "to where") is disclosed, never any data carried in the
-/// query. Fire-and-forget — never fails the caller.
+/// round-trip before it leaves the device. Fire-and-forget — never fails the
+/// caller.
 fn emit_backend_egress(path: &str) {
-    let endpoint = path.split('?').next().unwrap_or(path);
-    crate::openhuman::security::egress::emit_external_transfer(
-        crate::openhuman::security::egress::EgressDescriptor::integration(endpoint),
-    );
+    crate::openhuman::security::egress::emit_external_transfer(backend_egress_descriptor(path));
+}
+
+/// Local-only enforcement (privacy epic S7, #4441): refuse a managed-backend
+/// round-trip that ships user data when the live policy is `LocalOnly`. Returns
+/// `Ok(())` for control-plane paths (session / team / billing / integration
+/// connection-management + catalog) even under `LocalOnly` — blocking those
+/// would break sign-in and the Connections UI for no privacy benefit. See
+/// [`is_control_plane`](crate::openhuman::security::egress). Call before
+/// [`emit_backend_egress`] so a blocked call is neither disclosed nor sent.
+fn enforce_backend_egress(path: &str) -> anyhow::Result<()> {
+    crate::openhuman::security::egress::enforce_egress(&backend_egress_descriptor(path))
 }
 
 /// Shared client for all integration tools. Holds backend URL, auth token,
@@ -331,6 +346,7 @@ impl IntegrationClient {
         path: &str,
         body: &serde_json::Value,
     ) -> anyhow::Result<T> {
+        enforce_backend_egress(path)?;
         emit_backend_egress(path);
         self.ensure_budget_available(path).await?;
         let url = crate::api::config::api_url(&self.backend_url, path);
@@ -436,6 +452,7 @@ impl IntegrationClient {
 
     /// GET from a backend endpoint and parse the response `data` field.
     pub async fn get<T: serde::de::DeserializeOwned>(&self, path: &str) -> anyhow::Result<T> {
+        enforce_backend_egress(path)?;
         emit_backend_egress(path);
         self.ensure_budget_available(path).await?;
         let url = crate::api::config::api_url(&self.backend_url, path);
@@ -611,6 +628,8 @@ impl IntegrationClient {
         path: &str,
         body: &serde_json::Value,
     ) -> anyhow::Result<T> {
+        enforce_backend_egress(path)?;
+        emit_backend_egress(path);
         self.ensure_budget_available(path).await?;
         let url = crate::api::config::api_url(&self.backend_url, path);
         tracing::debug!("[integrations] PATCH {}", url);
@@ -632,6 +651,8 @@ impl IntegrationClient {
     /// Mirrors [`Self::post`] (auth header, 401 → session-expiry, error
     /// classification).
     pub async fn delete<T: serde::de::DeserializeOwned>(&self, path: &str) -> anyhow::Result<T> {
+        enforce_backend_egress(path)?;
+        emit_backend_egress(path);
         self.ensure_budget_available(path).await?;
         let url = crate::api::config::api_url(&self.backend_url, path);
         tracing::debug!("[integrations] DELETE {}", url);
@@ -656,6 +677,8 @@ impl IntegrationClient {
         path: &str,
         form: reqwest::multipart::Form,
     ) -> anyhow::Result<T> {
+        enforce_backend_egress(path)?;
+        emit_backend_egress(path);
         self.ensure_budget_available(path).await?;
         let url = crate::api::config::api_url(&self.backend_url, path);
         tracing::debug!("[integrations] POST(multipart) {}", url);
@@ -682,6 +705,8 @@ impl IntegrationClient {
         &self,
         path: &str,
     ) -> anyhow::Result<(bytes::Bytes, Option<String>, Option<String>)> {
+        enforce_backend_egress(path)?;
+        emit_backend_egress(path);
         self.ensure_budget_available(path).await?;
         let url = crate::api::config::api_url(&self.backend_url, path);
         tracing::debug!("[integrations] GET(bytes) {}", url);