feat(agent): record observed egress as http_request side effects (#465 Phase A, 3/5)

crates/flowproof-adapters/src/agent_runner.rs

@@ -668,13 +668,18 @@ pub fn run_against(
 /// other platform it is exactly [`run_against`] with an empty egress log,
 /// since the mechanism is Linux-only and the tier is reported "not
 /// contained" independently.
+///
+/// `egress_engaged` says whether the FLOW declared an egress policy, or is
+/// supervised for side-effect observation only under an allow-all set nobody
+/// declared - and the latter must never report `Enforced`.
 #[cfg(target_os = "linux")]
 pub fn run_against_contained(
     proxy: &AgentProxy,
     command: &str,
     env: &BTreeMap<String, String>,
     timeout: Duration,
     allow: &AllowSet,
+    egress_engaged: bool,
 ) -> Result<AgentRun, RunError> {
     let base = proxy.base_url();
     let mut cmd = configure(command, &base, env)?;
@@ -722,10 +727,16 @@ pub fn run_against_contained(
         // The filter that enforced is the filter that watched, so `fs`
         // above is evidence here and silence everywhere else.
         observed: true,
-        // Reaching here means the filter installed: it goes in via `pre_exec`
-        // and a failure aborts the spawn, so there is no path to a finished
-        // run with no filter behind it.
-        containment: Some(Containment::Enforced),
+        containment: Some(if egress_engaged {
+            // Reaching here means the filter installed (it goes in via
+            // `pre_exec`; a failure aborts the spawn) - and it enforced the
+            // DECLARED policy.
+            Containment::Enforced
+        } else {
+            // An allow-all policy nobody declared: observing side effects is
+            // not containing egress, and the tier must not blur the two.
+            Containment::observation_only()
+        }),
     };
     drop(log);
     Ok(run)
@@ -745,7 +756,13 @@ pub fn run_against_contained(
     env: &BTreeMap<String, String>,
     timeout: Duration,
     allow: &AllowSet,
+    egress_engaged: bool,
 ) -> Result<AgentRun, RunError> {
+    // Defense in depth: no caller passes `false` here today, but a future
+    // one must get the plain path, never WFP filters from a wildcard set.
+    if !egress_engaged {
+        return run_against(proxy, command, env, timeout);
+    }
     let command = command.trim();
     if command.is_empty() {
         return Err(RunError::NoCommand);
@@ -774,6 +791,7 @@ pub fn run_against_contained(
         egress: EgressLog {
             blocked: outcome.blocked,
             faults: outcome.faults,
+            observed: Vec::new(),
         },
         // Filesystem observation is a seccomp mechanism; Windows has none.
         fs: FsLog::default(),
@@ -794,9 +812,13 @@ pub fn run_against_contained(
     env: &BTreeMap<String, String>,
     timeout: Duration,
     _allow: &AllowSet,
+    egress_engaged: bool,
 ) -> Result<AgentRun, RunError> {
-    // No mechanism on this platform; the plain path, and the tier line says
-    // "not contained".
+    // Defense in depth, mirroring the Windows variant.
+    if !egress_engaged {
+        return run_against(proxy, command, env, timeout);
+    }
+    // No mechanism here; the plain path, and the tier says "not contained".
     run_against(proxy, command, env, timeout)
 }
 

crates/flowproof-adapters/src/egress.rs

@@ -56,6 +56,16 @@ impl Containment {
         )
     }
 
+    /// The tier for a run supervised for side-effect OBSERVATION only: an
+    /// allow-all policy nobody declared contains nothing.
+    pub fn observation_only() -> Self {
+        Containment::NotContained(
+            "flow engages side-effect observation only; no egress policy declared, \
+             nothing contained"
+                .to_string(),
+        )
+    }
+
     /// The tier for a `url:` flow: a service flowproof did not start cannot
     /// be contained.
     pub fn url_flow() -> Self {
@@ -124,6 +134,11 @@ impl Containment {
 pub struct EgressLog {
     /// Every denied attempt, in order - retries included.
     pub blocked: Vec<EgressEvent>,
+    /// Every ALLOWED non-loopback destination the supervisor performed a
+    /// connect/send to, in order - the `http_request` half of the side-effect
+    /// lane. Never folded into `blocked`: that list's emptiness is half the
+    /// `assert_no_egress` predicate.
+    pub observed: Vec<EgressEvent>,
     /// Every supervisor FAULT, in order. A fault is not a policy denial: it
     /// is a trapped syscall the supervisor could not adjudicate at all,
     /// because the mechanism it needs was refused (`process_vm_readv` or
@@ -190,6 +205,9 @@ pub struct AllowSet {
     /// Host entries are pre-resolved into `Ip` entries; only `Ip`/`Cidr`
     /// remain, each keeping its own optional port constraint.
     entries: Vec<AllowEntry>,
+    /// The spec spelling each entry came from, index-parallel with `entries`,
+    /// so [`AllowSet::allowed_as`] can name a destination by the spec's text.
+    spellings: Vec<String>,
 }
 
 impl AllowSet {
@@ -199,22 +217,43 @@ impl AllowSet {
     /// exempt). A name that does not resolve contributes no IPs, so its
     /// traffic is denied - the safe default.
     pub fn resolve(entries: &[String]) -> Result<Self, String> {
-        let mut out = Vec::new();
-        for raw in entries {
+        // The entry text is its own spelling when no unresolved form exists.
+        let pairs: Vec<(&str, &str)> = entries.iter().map(|e| (e.as_str(), e.as_str())).collect();
+        Self::resolve_spelled(&pairs)
+    }
+
+    /// `resolve`, from `(spelling, resolved)` pairs: the resolved text is
+    /// parsed and enforced; the spelling - the spec's own, possibly `${VAR}`,
+    /// text - is retained, so a destination such an entry admits can be
+    /// recorded by spelling rather than by the value it held.
+    pub fn resolve_spelled(pairs: &[(&str, &str)]) -> Result<Self, String> {
+        let mut set = Self::default();
+        for (spelling, raw) in pairs {
             let parsed = egress::parse_allow_entry(raw)?;
             match parsed.host {
                 HostMatch::Host(name) => {
                     for ip in resolve_host(&name) {
-                        out.push(AllowEntry {
+                        set.entries.push(AllowEntry {
                             host: HostMatch::Ip(ip),
                             port: parsed.port,
                         });
                     }
                 }
-                HostMatch::Ip(_) | HostMatch::Cidr(_, _) => out.push(parsed),
+                HostMatch::Ip(_) | HostMatch::Cidr(_, _) => set.entries.push(parsed),
             }
+            // One spelling per entry the pair contributed.
+            set.spellings
+                .resize(set.entries.len(), spelling.to_string());
         }
-        Ok(Self { entries: out })
+        Ok(set)
+    }
+
+    /// The observation-only policy: wildcard v4 + v6 CIDRs, so the POLICY
+    /// path admits every parseable inet destination. (The structural
+    /// refusals are independent of any set.)
+    pub fn allow_all() -> Self {
+        let all = ["0.0.0.0/0".to_string(), "::/0".to_string()];
+        Self::resolve(&all).expect("the wildcard CIDRs parse")
     }
 
     /// Is `(ip, port)` allowed? Loopback is allowed wholesale, independent of
@@ -229,6 +268,22 @@ impl AllowSet {
             .any(|entry| entry.port_ok(port) && host_matches(&entry.host, ip))
     }
 
+    /// The spec spelling under which `(ip, port)` is admitted, `None` when no
+    /// entry matches (loopback is exempt without consulting any entry). When
+    /// SEVERAL admit it, a `${VAR}` spelling wins over a concrete one.
+    pub fn allowed_as(&self, ip: IpAddr, port: u16) -> Option<&str> {
+        let mut concrete = None;
+        for (entry, spelling) in self.entries.iter().zip(&self.spellings) {
+            if entry.port_ok(port) && host_matches(&entry.host, ip) {
+                if spelling.contains("${") {
+                    return Some(spelling);
+                }
+                concrete.get_or_insert(spelling.as_str());
+            }
+        }
+        concrete
+    }
+
     /// No declared destinations: a contained run with an empty allow-set
     /// permits only loopback.
     pub fn is_empty(&self) -> bool {
@@ -336,6 +391,7 @@ mod tests {
     fn undeclared_destinations_dedupe_by_destination() {
         let log = EgressLog {
             faults: Vec::new(),
+            observed: Vec::new(),
             blocked: vec![
                 EgressEvent {
                     destination: "198.51.100.9:443".into(),
@@ -367,6 +423,7 @@ mod tests {
     fn a_faulted_log_is_not_clean_even_with_nothing_blocked() {
         let log = EgressLog {
             blocked: Vec::new(),
+            observed: Vec::new(),
             faults: vec!["connect: process_vm_readv: Operation not permitted".into()],
         };
         // The old predicate - "nothing was blocked" - still reads as empty.
@@ -379,6 +436,7 @@ mod tests {
     fn repeated_faults_dedupe_but_keep_first_seen_order() {
         let log = EgressLog {
             blocked: Vec::new(),
+            observed: Vec::new(),
             faults: vec![
                 "connect: process_vm_readv: EPERM".into(),
                 "sendto: pidfd_getfd: EPERM".into(),

crates/flowproof-adapters/src/egress_linux.rs

@@ -768,6 +768,7 @@ fn handle_connect(
             let ip = normalize(ip);
             if allow.allows(ip, port) {
                 dbg_egress(&format!("connect: ALLOWED {ip}:{port}, performing"));
+                record_observed(log, spawn, ip, port, "tcp");
                 perform_connect(req.pid, sockfd, &buf[..n], log)
             } else {
                 dbg_egress(&format!("connect: DENIED by policy {ip}:{port}"));
@@ -830,6 +831,7 @@ fn handle_sendto(
                     record(log, spawn, &format!("{ip}:{port}"), "udp");
                     return errno_resp(libc::ECONNREFUSED);
                 }
+                record_observed(log, spawn, ip, port, "udp");
                 Some((addr, n))
             }
             Some(Dest::Unix) => Some((addr, n)),
@@ -875,6 +877,7 @@ fn handle_sendmsg(
     }
     let (name_ptr, name_len, control_len) = msghdr_fields(&hdr);
 
+    let mut vetted = None;
     if name_ptr != 0 && name_len != 0 {
         let mut addr = [0u8; 128];
         let read = read_child(req.pid, name_ptr, &mut addr, name_len as usize);
@@ -894,13 +897,18 @@ fn handle_sendmsg(
                 record(log, spawn, &format!("{ip}:{port}"), "udp");
                 return errno_resp(libc::ECONNREFUSED);
             }
+            vetted = Some((ip, port));
         }
     }
     // Control data (e.g. SCM_RIGHTS) is not marshalled in v1: refuse rather
     // than forward something we did not fully copy.
     if control_len != 0 {
         return errno_resp(libc::EPERM);
     }
+    // AFTER the refusal, so a refused message is never reported as performed.
+    if let Some((ip, port)) = vetted {
+        record_observed(log, spawn, ip, port, "udp");
+    }
     // On-host / connected: re-perform on the child's own socket. sendmmsg is
     // serviced as its first message; the child sees one message sent.
     perform_sendmsg(req.pid, sockfd, msg_ptr, log)
@@ -1518,6 +1526,26 @@ fn record(log: &Mutex<EgressLog>, spawn: Instant, destination: &str, protocol: &
         .push(event);
 }
 
+/// Record an ALLOWED destination into the shared log's `observed` list - the
+/// `http_request` half of the side-effect lane. Loopback is filtered HERE, at
+/// the capture point: connects to flowproof's own proxy and MCP listeners are
+/// already first-class trace data, and re-emitting them would double-count
+/// flowproof's boundaries as agent behavior (and pay a push per model call).
+fn record_observed(log: &Mutex<EgressLog>, spawn: Instant, ip: IpAddr, port: u16, protocol: &str) {
+    if is_loopback(ip) {
+        return;
+    }
+    let event = EgressEvent {
+        destination: format!("{ip}:{port}"),
+        protocol: protocol.to_string(),
+        at_ms: spawn.elapsed().as_millis() as u64,
+    };
+    log.lock()
+        .unwrap_or_else(|e| e.into_inner())
+        .observed
+        .push(event);
+}
+
 /// Record a supervisor FAULT and deny the syscall.
 ///
 /// A fault is not a denial. It is the supervisor admitting it could not
@@ -1804,4 +1832,46 @@ mod tests {
         assert_eq!(ok_resp(5).val, 5);
         assert_eq!(ok_resp(5).error, 0);
     }
+
+    /// The narrowed observation-only pin: the POLICY path under allow-all
+    /// admits every parseable inet destination - deliberately NOT "denies
+    /// nothing", since the structural refusals are independent of any
+    /// `AllowSet`. The capture point records an allowed destination into
+    /// `observed` - never `blocked` - loopback filtered wholesale, and the
+    /// tier such a run reports never says enforced.
+    #[test]
+    fn allow_all_admits_every_parseable_inet_destination_and_observation_records_it() {
+        let all = AllowSet::allow_all();
+        assert!(!all.is_empty());
+        let destinations: [(IpAddr, u16); 4] = [
+            ("198.51.100.9".parse().expect("ipv4"), 443),
+            ("10.1.2.3".parse().expect("ipv4"), 8080),
+            ("2001:db8::1".parse().expect("ipv6"), 65535),
+            // v4-mapped, as `normalize` would hand it to the policy.
+            (normalize("::ffff:203.0.113.9".parse().expect("ipv6")), 53),
+        ];
+        for (ip, port) in destinations {
+            assert!(all.allows(ip, port), "{ip}:{port} must be admitted");
+        }
+
+        let log = Mutex::new(EgressLog::default());
+        let spawn = Instant::now();
+        let tcp_dest: IpAddr = "198.51.100.9".parse().expect("ipv4");
+        record_observed(&log, spawn, tcp_dest, 443, "tcp");
+        record_observed(&log, spawn, "203.0.113.9".parse().expect("ipv4"), 53, "udp");
+        for loopback in ["127.0.0.1", "127.9.9.9", "::1"] {
+            record_observed(&log, spawn, loopback.parse().expect("loopback"), 443, "tcp");
+        }
+        let log = log.into_inner().expect("unpoisoned");
+        assert!(log.blocked.is_empty(), "observation is not a denial");
+        assert_eq!(log.observed.len(), 2, "loopback never appears");
+        assert_eq!(log.observed[0].destination, "198.51.100.9:443");
+        assert_eq!(log.observed[0].protocol, "tcp");
+        assert_eq!(log.observed[1].destination, "203.0.113.9:53");
+        assert_eq!(log.observed[1].protocol, "udp");
+
+        let line = Containment::observation_only().report_line();
+        assert!(line.contains("not contained"), "{line}");
+        assert!(!line.contains("enforced"), "{line}");
+    }
 }