sandbox,server: surface per-path L7 escalations as fresh draft chunks

crates/openshell-server/src/persistence/postgres.rs

@@ -736,10 +736,15 @@ ORDER BY created_at_ms DESC
         }
         let payload = draft_chunk_payload_from_record(&record)?;
 
+        // Clear the dedup target once a chunk is decided: new observations for
+        // the same host|port|binary must surface as a fresh pending chunk
+        // (possibly carrying new L7 evidence) instead of silently folding
+        // their hit_count into a row the reviewer already acted on.
         let result = sqlx::query(
             r"
 UPDATE objects
-SET status = $3, payload = $4, updated_at_ms = $5
+SET status = $3, payload = $4, updated_at_ms = $5,
+    dedup_key = CASE WHEN $3 = 'pending' THEN dedup_key ELSE NULL END
 WHERE object_type = $1 AND id = $2
 ",
         )

crates/openshell-server/src/persistence/sqlite.rs

@@ -758,10 +758,15 @@ ORDER BY "created_at_ms" DESC
         }
         let payload = draft_chunk_payload_from_record(&record)?;
 
+        // Clear the dedup target once a chunk is decided: new observations for
+        // the same host|port|binary must surface as a fresh pending chunk
+        // (possibly carrying new L7 evidence) instead of silently folding
+        // their hit_count into a row the reviewer already acted on.
         let result = sqlx::query(
             r#"
 UPDATE "objects"
-SET "status" = ?3, "payload" = ?4, "updated_at_ms" = ?5
+SET "status" = ?3, "payload" = ?4, "updated_at_ms" = ?5,
+    "dedup_key" = CASE WHEN ?3 = 'pending' THEN "dedup_key" ELSE NULL END
 WHERE "object_type" = ?1 AND "id" = ?2
 "#,
         )

crates/openshell-supervisor-network/src/l7/graphql.rs

@@ -804,6 +804,7 @@ network_policies:
             activity_tx: None,
             dynamic_credentials: None,
             token_grant_resolver: None,
+            denial_tx: None,
         };
         let request_info = crate::l7::L7RequestInfo {
             action: req.action,

crates/openshell-supervisor-network/src/l7/relay.rs

@@ -13,6 +13,7 @@ use crate::l7::{EnforcementMode, L7EndpointConfig, L7Protocol, L7RequestInfo};
 use crate::opa::{PolicyGenerationGuard, TunnelPolicyEngine};
 use miette::{IntoDiagnostic, Result, miette};
 use openshell_core::activity::{ActivitySender, try_record_activity};
+use openshell_core::denial::DenialEvent;
 use openshell_core::secrets::{self, SecretResolver};
 use openshell_ocsf::{
     ActionId, ActivityId, DispositionId, Endpoint, HttpActivityBuilder, HttpRequest,
@@ -51,6 +52,10 @@ pub struct L7EvalContext {
     /// Dynamic token grant resolver for endpoint-bound credentials.
     pub(crate) token_grant_resolver:
         Option<Arc<dyn crate::l7::token_grant_injection::TokenGrantResolver>>,
+    /// Denial aggregator channel. L7 request denials feed the same
+    /// observation-driven policy analysis as connect-stage denials, carrying
+    /// the observed method/path so proposals can be path-aware.
+    pub(crate) denial_tx: Option<tokio::sync::mpsc::UnboundedSender<DenialEvent>>,
 }
 
 #[derive(Default)]
@@ -464,6 +469,28 @@ fn emit_l7_request_log(
         .build();
     ocsf_emit!(event);
     emit_activity(ctx, decision_str == "deny", "l7_policy");
+    if decision_str == "deny" {
+        emit_l7_denial(ctx, request_info, redacted_target, reason);
+    }
+}
+
+/// Feed an L7 request denial to the denial aggregator (if configured) so the
+/// observation-driven analysis can propose path-aware rules. The target is
+/// already redacted (no query string / credentials), matching what the OCSF
+/// log records.
+fn emit_l7_denial(ctx: &L7EvalContext, request_info: &L7RequestInfo, path: &str, reason: &str) {
+    if let Some(tx) = &ctx.denial_tx {
+        let _ = tx.send(DenialEvent {
+            host: ctx.host.clone(),
+            port: ctx.port,
+            binary: ctx.binary_path.clone(),
+            ancestors: ctx.ancestors.clone(),
+            deny_reason: reason.to_string(),
+            denial_stage: "l7".to_string(),
+            l7_method: Some(request_info.action.clone()),
+            l7_path: Some(path.to_string()),
+        });
+    }
 }
 
 fn emit_activity(ctx: &L7EvalContext, denied: bool, deny_group: &'static str) {
@@ -774,6 +801,9 @@ where
                 ))
                 .build();
             ocsf_emit!(event);
+            if decision_str == "deny" {
+                emit_l7_denial(ctx, &request_info, &redacted_target, &reason);
+            }
         }
 
         // Store the resolved target for the deny response redaction
@@ -1424,6 +1454,7 @@ network_policies:
             activity_tx: None,
             dynamic_credentials: Some(fixture.dynamic_credentials()),
             token_grant_resolver: Some(fixture.resolver()),
+            denial_tx: None,
         };
 
         (config, tunnel_engine, ctx, fixture)
@@ -1467,6 +1498,7 @@ network_policies:
             activity_tx: None,
             dynamic_credentials: Some(fixture.dynamic_credentials()),
             token_grant_resolver: Some(fixture.resolver()),
+            denial_tx: None,
         };
 
         (generation_guard, ctx, fixture)
@@ -1482,6 +1514,53 @@ network_policies:
             .count()
     }
 
+    /// An L7 deny must feed the denial aggregator with the observed
+    /// method/path so observation-driven analysis can propose path-aware
+    /// rules; allows must not.
+    #[test]
+    fn l7_deny_emits_denial_event_with_method_and_path() {
+        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
+        let ctx = L7EvalContext {
+            host: "api.example.test".into(),
+            port: 443,
+            policy_name: "rest_api".into(),
+            binary_path: "/usr/bin/gh".into(),
+            ancestors: vec!["/bin/bash".into()],
+            cmdline_paths: vec![],
+            secret_resolver: None,
+            activity_tx: None,
+            dynamic_credentials: None,
+            token_grant_resolver: None,
+            denial_tx: Some(tx),
+        };
+        let request = L7RequestInfo {
+            action: "GET".into(),
+            target: "/user".into(),
+            query_params: std::collections::HashMap::new(),
+            graphql: None,
+        };
+
+        emit_l7_request_log(
+            &ctx,
+            &request,
+            "/user",
+            "deny",
+            "l7",
+            "GET /user not permitted by policy",
+            None,
+        );
+        let event = rx.try_recv().expect("deny must emit a denial event");
+        assert_eq!(event.host, "api.example.test");
+        assert_eq!(event.port, 443);
+        assert_eq!(event.binary, "/usr/bin/gh");
+        assert_eq!(event.denial_stage, "l7");
+        assert_eq!(event.l7_method.as_deref(), Some("GET"));
+        assert_eq!(event.l7_path.as_deref(), Some("/user"));
+
+        emit_l7_request_log(&ctx, &request, "/user", "allow", "l7", "", None);
+        assert!(rx.try_recv().is_err(), "allow must not emit a denial event");
+    }
+
     #[test]
     fn parse_rejection_detail_adds_l7_hint_for_encoded_slash() {
         let detail = parse_rejection_detail(
@@ -1786,6 +1865,7 @@ network_policies:
             activity_tx: None,
             dynamic_credentials: None,
             token_grant_resolver: None,
+            denial_tx: None,
         };
         let request = L7RequestInfo {
             action: "WEBSOCKET_TEXT".into(),
@@ -1844,6 +1924,7 @@ network_policies:
             activity_tx: None,
             dynamic_credentials: None,
             token_grant_resolver: None,
+            denial_tx: None,
         };
 
         let (mut app, mut relay_client) = tokio::io::duplex(8192);
@@ -1951,6 +2032,7 @@ network_policies:
             activity_tx: None,
             dynamic_credentials: None,
             token_grant_resolver: None,
+            denial_tx: None,
         };
 
         let (mut app, mut relay_client) = tokio::io::duplex(8192);
@@ -2071,6 +2153,7 @@ network_policies:
             activity_tx: None,
             dynamic_credentials: None,
             token_grant_resolver: None,
+            denial_tx: None,
         };
 
         let (mut app, mut relay_client) = tokio::io::duplex(8192);
@@ -2244,6 +2327,7 @@ network_policies:
             activity_tx: None,
             dynamic_credentials: None,
             token_grant_resolver: None,
+            denial_tx: None,
         };
 
         let (mut app, mut relay_client) = tokio::io::duplex(8192);
@@ -2334,6 +2418,7 @@ network_policies:
             activity_tx: None,
             dynamic_credentials: None,
             token_grant_resolver: None,
+            denial_tx: None,
         };
 
         let (mut app, mut relay_client) = tokio::io::duplex(8192);

crates/openshell-supervisor-network/src/l7/token_grant_injection.rs

@@ -735,6 +735,7 @@ mod tests {
             activity_tx: None,
             dynamic_credentials: Some(fixture.dynamic_credentials()),
             token_grant_resolver: Some(fixture.resolver()),
+            denial_tx: None,
         };
         let req = L7Request {
             action: "GET".to_string(),
@@ -772,6 +773,7 @@ mod tests {
             activity_tx: None,
             dynamic_credentials: Some(fixture.dynamic_credentials()),
             token_grant_resolver: Some(fixture.resolver()),
+            denial_tx: None,
         };
         let req = L7Request {
             action: "GET".to_string(),

crates/openshell-supervisor-network/src/l7/websocket.rs

@@ -1273,6 +1273,7 @@ network_policies:
             activity_tx: None,
             dynamic_credentials: None,
             token_grant_resolver: None,
+            denial_tx: None,
         };
         let (mut client_write, mut relay_read) = tokio::io::duplex(MAX_TEXT_MESSAGE_BYTES + 1024);
         let (mut relay_write, mut upstream_read) = tokio::io::duplex(MAX_TEXT_MESSAGE_BYTES + 1024);

crates/openshell-supervisor-network/src/proxy.rs

@@ -970,6 +970,7 @@ async fn handle_tcp_connection(
         token_grant_resolver: dynamic_credentials
             .as_ref()
             .map(|_| crate::l7::token_grant_injection::default_resolver()),
+        denial_tx: denial_tx.clone(),
     };
 
     if effective_tls_skip {
@@ -3215,6 +3216,7 @@ async fn handle_forward_proxy(
         token_grant_resolver: dynamic_credentials
             .as_ref()
             .map(|_| crate::l7::token_grant_injection::default_resolver()),
+        denial_tx: denial_tx.cloned(),
     };
     let mut l7_activity_pending = false;
 
@@ -4293,6 +4295,7 @@ mod tests {
             activity_tx: None,
             dynamic_credentials: Some(fixture.dynamic_credentials()),
             token_grant_resolver: Some(fixture.resolver()),
+            denial_tx: None,
         };
 
         (ctx, fixture)
@@ -4351,6 +4354,7 @@ mod tests {
             activity_tx: None,
             dynamic_credentials: None,
             token_grant_resolver: None,
+            denial_tx: None,
         };
         (config, tunnel_engine, ctx)
     }
@@ -4519,6 +4523,7 @@ mod tests {
             activity_tx: None,
             dynamic_credentials: None,
             token_grant_resolver: None,
+            denial_tx: None,
         };
         let query_params = std::collections::HashMap::new();
 
@@ -4562,6 +4567,7 @@ mod tests {
             activity_tx: None,
             dynamic_credentials: None,
             token_grant_resolver: None,
+            denial_tx: None,
         };
         let query_params = std::collections::HashMap::new();
         let config = websocket_l7_config(crate::l7::L7Protocol::Rest, false);