rnet: export listener & summary via pkg backend

Author: ignoramous

intra/backend/rnet_services.go

@@ -53,3 +53,28 @@ type Services interface {
 	// Refresh re-registers servces and returns a csv of active ones.
 	RefreshServers() (active string)
 }
+
+// Summary is a summary of a DNS transaction, reported when it is complete.
+type ServerSummary struct {
+	Type     string // http1, socks5, etc.
+	SID      string // Server ID.
+	PID      string // Proxy ID (hop) that handled egress, if any.
+	CID      string // Connection id
+	Tx       int64  // Total uploaded (bytes).
+	Rx       int64  // Total downloaded (bytes).
+	Duration int64  // Conn open duration (millis).
+	Msg      string // Error message, if any.
+}
+
+// ServerListener receives Server events.
+type ServerListener interface {
+	// SvcRoute decides how to forward an incoming connection over service (sid).
+	SvcRoute(sid, pid, network, sipport, dipport string) *Tab
+	// OnSvcComplete reports summary after a connection closes.
+	OnSvcComplete(*ServerSummary)
+}
+
+type Tab struct {
+	CID   string // CID is the ID of this connection.
+	Block bool   // Block is true if this connection should be blocked.
+}

intra/dns53/dot_test.go

@@ -25,7 +25,6 @@ import (
 	"github.com/celzero/firestack/intra/ipn/warp"
 	ilog "github.com/celzero/firestack/intra/log"
 	"github.com/celzero/firestack/intra/protect"
-	"github.com/celzero/firestack/intra/rnet"
 	"github.com/celzero/firestack/intra/settings"
 	"github.com/celzero/firestack/intra/x64"
 	"github.com/celzero/firestack/intra/xdns"
@@ -111,7 +110,7 @@ type fakeBdg struct {
 
 var (
 	// baseNsOpts = &x.DNSOpts{PIDCSV: dnsx.NetBaseProxy, IPCSV: "", TIDCSV: x.CT + "test0"}
-	baseTab  = &rnet.Tab{CID: "testcid", Block: false}
+	baseTab  = &x.Tab{CID: "testcid", Block: false}
 	seNsOpts = &x.DNSOpts{PIDCSV: ipn.RpnSE, IPCSV: "", TIDCSV: x.CT + "test0"}
 )
 
@@ -121,8 +120,8 @@ func (*fakeBdg) OnDNSAdded(string)                     {}
 func (*fakeBdg) OnDNSRemoved(string)                   {}
 func (*fakeBdg) OnDNSStopped()                         {}
 
-func (*fakeBdg) Route(a, b, c, d, e string) *rnet.Tab { return baseTab }
-func (*fakeBdg) OnComplete(*rnet.ServerSummary)       {}
+func (*fakeBdg) Route(a, b, c, d, e string) *x.Tab { return baseTab }
+func (*fakeBdg) OnComplete(*x.ServerSummary)       {}
 
 const minmtu = 1280
 const dualstack = settings.IP46

intra/rnet/http.go

@@ -138,7 +138,7 @@ func (h *httpx) route(req *http.Request, ctx *tx.ProxyCtx) (*http.Request, *http
 	src := req.RemoteAddr
 	sid := h.id
 	pid := h.pid()
-	tab := h.listener.Route(sid, pid, "tcp", src, req.Host)
+	tab := h.listener.SvcRoute(sid, pid, "tcp", src, req.Host)
 	log.D("svchttp: route: tab(%v) id(%s) p(%s) src(%s) dst(%s)", tab, h.id, pid, src, req.Host)
 	if tab.Block {
 		return req, tx.NewResponse(req, tx.ContentTypeText, http.StatusForbidden, "Forbidden")
@@ -161,12 +161,12 @@ func (h *httpx) summarize(res *http.Response, ctx *tx.ProxyCtx) *http.Response {
 		log.W("svchttp: summarize: invalid userdata %v", ctx.UserData)
 		return res
 	}
-	ssu.Rx = int(res.ContentLength)
+	ssu.Rx = res.ContentLength
 	if req != nil {
-		ssu.Tx = int(req.ContentLength)
+		ssu.Tx = req.ContentLength
 	}
 	ssu.done(errNop)
-	go h.listener.OnComplete(ssu)
+	go h.listener.OnSvcComplete(ssu.ServerSummary)
 	return res
 }
 
@@ -175,7 +175,7 @@ func (h *httpx) routeConnect(host string, ctx *tx.ProxyCtx) (*tx.ConnectAction,
 	dst := ctx.Req.Host
 	sid := h.id
 	pid := h.pid()
-	tab := h.listener.Route(sid, pid, "tcp", src, host)
+	tab := h.listener.SvcRoute(sid, pid, "tcp", src, host)
 	log.D("svchttp: routeConnect: tab(%v) id(%s) p(%s) src(%s) dst(%s)", tab, h.id, pid, src, dst)
 	if tab.Block {
 		return tx.RejectConnect, host
@@ -224,7 +224,7 @@ func (h *httpx) hijackConnect(req *http.Request, client net.Conn, ctx *tx.ProxyC
 			wg.Wait()
 			clos(client, target)
 		}
-		h.listener.OnComplete(ssu)
+		h.listener.OnSvcComplete(ssu.ServerSummary)
 	}()
 }
 

intra/rnet/listener.go

@@ -11,30 +11,23 @@ import (
 	"fmt"
 	"time"
 
+	x "github.com/celzero/firestack/intra/backend"
 	"github.com/celzero/firestack/intra/core"
 )
 
 var errNop = errors.New("no error")
 
-// Summary is a summary of a DNS transaction, reported when it is complete.
 type ServerSummary struct {
-	Type     string    // http1, socks5, etc.
-	SID      string    // Server id
-	PID      string    // Proxy ID (hop) that handled egress, if any.
-	CID      string    // Connection id
-	Tx       int       // Amount uploaded (bytes).
-	Rx       int       // Amount downloaded (bytes).
-	Duration int32     // Conn open duration (seconds).
-	start    time.Time // Tracks start time; unexported.
-	Msg      string    // Error message, if any.
+	*x.ServerSummary
+	start time.Time // Tracks start time; unexported.
 }
 
 func (s *ServerSummary) done(errs ...error) {
 	if s == nil {
 		return
 	}
 
-	s.Duration = int32(time.Since(s.start).Seconds())
+	s.Duration = time.Since(s.start).Milliseconds()
 
 	err := core.JoinErr(errs...) // errs may be nil
 	if err != nil {
@@ -59,24 +52,13 @@ func (s *ServerSummary) String() string {
 
 func serverSummary(typ, sid, pid, cid string) *ServerSummary {
 	return &ServerSummary{
-		Type:  typ,
-		SID:   sid,
-		PID:   pid,
-		CID:   cid,
+		ServerSummary: &x.ServerSummary{
+			Type: typ,
+			SID:  sid,
+			PID:  pid,
+			CID:  cid,
+			Msg:  errNop.Error(),
+		},
 		start: time.Now(),
-		Msg:   errNop.Error(),
 	}
 }
-
-// ServerListener receives Server events.
-type ServerListener interface {
-	// Route decides how to forward an incoming connection over service (sid).
-	Route(sid, pid, network, sipport, dipport string) *Tab
-	// OnComplete reports summary after a connection closes.
-	OnComplete(*ServerSummary)
-}
-
-type Tab struct {
-	CID   string // CID is the ID of this connection.
-	Block bool   // Block is true if this connection should be blocked.
-}

intra/rnet/servers.go

@@ -49,10 +49,12 @@ var (
 
 // todo: github.com/txthinking/brook/blob/master/pac.go
 
-type Server = x.Server
+type Server x.Server
 
 type Services x.Services
 
+type ServerListener x.ServerListener
+
 var _ Services = (*services)(nil)
 var _ Server = (*httpx)(nil)
 var _ Server = (*socks5)(nil)
@@ -79,7 +81,7 @@ func NewServices(pctx context.Context, proxies ipn.Proxies, ctl protect.Controll
 	return svc
 }
 
-func (s *services) AddServer(id, url string) (svc Server, err error) {
+func (s *services) AddServer(id, url string) (svc x.Server, err error) {
 	s.RemoveServer(id)
 
 	switch id {
@@ -146,7 +148,7 @@ func (s *services) RemoveServer(id string) bool {
 	return false
 }
 
-func (s *services) GetServer(id string) (Server, error) {
+func (s *services) GetServer(id string) (x.Server, error) {
 	s.RLock()
 	defer s.RUnlock()
 

intra/rnet/socks5.go

@@ -237,8 +237,8 @@ func (h *socks5) pid() (x string) {
 	return
 }
 
-func (h *socks5) route(network, src, dst string) *Tab {
-	return h.listener.Route(h.id, h.pid(), network, src, dst)
+func (h *socks5) route(network, src, dst string) *x.Tab {
+	return h.listener.SvcRoute(h.id, h.pid(), network, src, dst)
 }
 
 func (h *socks5) candial() error {
@@ -260,7 +260,7 @@ func (h *socks5) setDeadline(c net.Conn, secs int) error {
 }
 
 type pipefin struct {
-	ex  int   // bytes exchanged
+	ex  int64 // bytes exchanged
 	err error // error, if any
 }
 
@@ -272,7 +272,7 @@ func (h *socks5) pipe(r, w net.Conn, finch chan<- pipefin) {
 		*bptr = bf
 		core.Recycle(bptr)
 	}()
-	ex := 0
+	ex := int64(0)
 	laddr := r.LocalAddr()
 	raddr := w.RemoteAddr()
 	for {
@@ -281,7 +281,7 @@ func (h *socks5) pipe(r, w net.Conn, finch chan<- pipefin) {
 			break
 		}
 		n, err := r.Read(bf[:])
-		ex += n
+		ex += int64(n)
 		if err != nil {
 			log.E("svcsocks5: tcp: %s; read %s; err: %v", h.ID(), laddr, err)
 			finch <- pipefin{ex, err}
@@ -304,10 +304,10 @@ func (h *socks5) tcphandle(s *tx.Server, ingress *net.TCPConn, r *tx.Request) (e
 		var cid string
 		var egress *net.TCPConn
 		cid, egress, err = h.Connect(r, ingress)
-		summary := serverSummary(h.Type(), h.ID(), h.pid(), cid)
+		ssu := serverSummary(h.Type(), h.ID(), h.pid(), cid)
 		defer func() {
-			summary.done(err)
-			go h.listener.OnComplete(summary)
+			ssu.done(err)
+			go h.listener.OnSvcComplete(ssu.ServerSummary)
 		}()
 
 		log.D("svcsocks5: proxy-tcp: %s; socks5-connect %s", cid, r.Address())
@@ -329,8 +329,8 @@ func (h *socks5) tcphandle(s *tx.Server, ingress *net.TCPConn, r *tx.Request) (e
 
 		err = core.JoinErr(finrx.err, fintx.err)
 
-		summary.Rx = finrx.ex
-		summary.Tx = fintx.ex
+		ssu.Rx = finrx.ex
+		ssu.Tx = fintx.ex
 
 		return err
 	}
@@ -386,7 +386,7 @@ func (h *socks5) udphandle(s *tx.Server, addr *net.UDPAddr, pkt *tx.Datagram) (e
 			// writing to egress conn
 			n, werr := egress.RemoteConn.Write(data)
 			if ssu != nil {
-				ssu.Tx += n
+				ssu.Tx += int64(n)
 			}
 			log.D("svcsocks5: udp: %s; data sent; (err: %v / summary? %t)? client: %s server: %s remote: %s sz: %d", cid, werr, ssu != nil, uecaddr, ueladdr, ueraddr, n)
 			if werr != nil {
@@ -409,7 +409,7 @@ func (h *socks5) udphandle(s *tx.Server, addr *net.UDPAddr, pkt *tx.Datagram) (e
 	ssu := serverSummary(h.Type(), h.ID(), h.pid(), cid)
 	defer func() {
 		ssu.done(err)
-		go h.listener.OnComplete(ssu)
+		go h.listener.OnSvcComplete(ssu.ServerSummary)
 	}()
 
 	log.D("svcsocks5: udp: %s; dst %s", cid, dst)
@@ -473,7 +473,7 @@ func (h *socks5) udphandle(s *tx.Server, addr *net.UDPAddr, pkt *tx.Datagram) (e
 					log.E("svcsocks5: udp: %s; read err: %v", cid, err)
 					return
 				}
-				ssu.Rx += n
+				ssu.Rx += int64(n)
 				log.D("svcsocks5: udp: %s; got data; client: %s server: %s remote: %s data: %d", cid, uecaddr, ueladdr, ueraddr, n)
 				a, addr, port, err := tx.ParseAddress(dst)
 				if err != nil {