tcp,udp,icmp: impl post flow

Author: ignoramous

intra/common.go

@@ -60,6 +60,7 @@ var _ SocketListener = (*zeroListener)(nil)
 func (*zeroListener) Preflow(_, _ int32, _, _ *x.Gostr) *PreMark       { return nil }
 func (*zeroListener) Flow(_, _ int32, _, _, _, _, _, _ *x.Gostr) *Mark { return nil }
 func (*zeroListener) Inflow(_, _ int32, _, _ *x.Gostr) *Mark           { return nil }
+func (*zeroListener) PostFlow(*Mark)                                   {}
 func (*zeroListener) OnSocketClosed(*SocketSummary)                    {}
 
 var nooplistener = new(zeroListener)

intra/icmp.go

@@ -115,6 +115,8 @@ func (h *icmpHandler) Ping(msg []byte, source, target netip.AddrPort) (echoed bo
 	h.conntracker.Track(cid, uid, pidstr(px), uc)
 	defer h.conntracker.Untrack(cid)
 
+	h.listener.PostFlow(smm.postMark())
+
 	tx = len(msg)
 	// todo: construct ICMP header? github.com/prometheus-community/pro-bing/blob/0bacb2d5e7/ping.go#L717
 	reply, from, err := core.Echo(uc, msg, net.UDPAddrFromAddrPort(dst), target.Addr().Is4())

intra/listener.go

@@ -68,6 +68,10 @@ type SocketListener interface {
 	// except for the CID field, which is sent back via OnSocketClosed, and "Block" proxy which
 	// will drop this connection on the floor.
 	Inflow(protocol, uid int32, src, dst *x.Gostr) *Mark
+	// PostFlow is called after a flow is marked by Flow or Inflow.
+	// It denotes the final Mark that was applied to the flow.
+	// The only major discernable effect is PIDCSV has a single PID.
+	PostFlow(m *Mark)
 	// OnSocketClosed reports summary after a socket closes.
 	OnSocketClosed(*SocketSummary)
 }
@@ -135,6 +139,17 @@ func udpSummary(id, uid string, dst netip.Addr) *SocketSummary {
 	return s
 }
 
+func (s *SocketSummary) postMark() *Mark {
+	if s == nil {
+		return nil
+	}
+	return &Mark{
+		PIDCSV: s.PID,
+		CID:    s.ID,
+		UID:    s.UID,
+	}
+}
+
 // String implements fmt.Stringer.
 func (s *SocketSummary) String() string {
 	if s != nil {

intra/tcp.go

@@ -289,6 +289,7 @@ func (h *tcpHandler) handle(px ipn.Proxy, src net.Conn, boundSrc, target netip.A
 	}
 
 	core.Go("tcp.forward."+smm.ID, func() {
+		h.listener.PostFlow(smm.postMark())
 		h.forward(src, rwext{dst, tcptimeout}, smm) // src always *gonet.TCPConn
 	})
 	return nil // handled; takes ownership of src

intra/udp.go

@@ -169,6 +169,7 @@ func (h *udpHandler) proxy(gconn *netstack.GUDPConn, src, dst netip.AddrPort, dm
 
 	cid := smm.ID
 	core.Go("udp.forward."+cid, func() {
+		h.listener.PostFlow(smm.postMark())
 		h.forward(gconn, rwext{remote, udptimeout}, smm)
 	})
 	return true // ok