From 41549b64a53ea0f7066f19eb7b2d1a2437d9957f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E6=80=9D=E6=95=8F?= Date: Wed, 31 Aug 2022 16:28:43 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config.go | 2 +- go.mod | 6 +- go.sum | 4 + main.go | 4 +- netcard.go | 327 ++++++++++++++--------------------------------------- rpc.go | 56 --------- 6 files changed, 99 insertions(+), 300 deletions(-) delete mode 100644 rpc.go diff --git a/config.go b/config.go index 3dbfec8..49fd6b8 100644 --- a/config.go +++ b/config.go @@ -50,7 +50,7 @@ func NewConfig() *Config { log.Panic(err) } - cnf.IPv4Nodes[binary.LittleEndian.Uint32([]byte(inet))] = node.Real + cnf.IPv4Nodes[binary.LittleEndian.Uint32([]byte(inet.To4()))] = node.Real log.Printf("real %v virt %v %v", node.Real, inet, net) } diff --git a/go.mod b/go.mod index 3eebf69..269d83a 100644 --- a/go.mod +++ b/go.mod @@ -4,9 +4,13 @@ go 1.19 require github.com/songgao/water v0.0.0-20200317203138-2b4b6d7c09d8 -require github.com/klauspost/compress v1.15.9 // indirect +require ( + github.com/474420502/check v0.0.1 // indirect + github.com/klauspost/compress v1.15.9 // indirect +) require ( + github.com/474420502/perfectshutdown v0.3.1 github.com/474420502/random v0.5.1 github.com/golang/protobuf v1.5.2 // indirect github.com/songgao/packets v0.0.0-20160404182456-549a10cd4091 // indirect diff --git a/go.sum b/go.sum index 5d4f5c3..ff734c2 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,9 @@ cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw= cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw= +github.com/474420502/check v0.0.1 h1:JliRiu5sLCd5Xo3Dh0lZqVT/3M8iHRNm/wkjgM3s/wk= +github.com/474420502/check v0.0.1/go.mod h1:W4LYCKwmpJsZMfWy24YrlC18bV7tuJVhiWHkXSAq1Bg= +github.com/474420502/perfectshutdown v0.3.1 h1:HFWr5pfcKOHnEe39B5PFKw2ipHCr3jKDPEvuTFMHqeo= +github.com/474420502/perfectshutdown v0.3.1/go.mod h1:qonD0eLCz1ncQq8heGQHbPdLlh30lofF2ISzV5ASW8A= github.com/474420502/random v0.5.1 h1:AA/r690QEKiVdCsFD9OKbjUmmtmLJQ7Qxd4IMVV1sbc= github.com/474420502/random v0.5.1/go.mod h1:Kq5VC8ezVvzp9CUahvebIdlSiFdHzEzscaOScLvbw2c= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= diff --git a/main.go b/main.go index ea6ee86..751b2e2 100644 --- a/main.go +++ b/main.go @@ -19,8 +19,8 @@ func init() { //go:generate bash -c "protoc --go_out=plugins=grpc:. proto/*.proto" func main() { - testmain() - // NewNetCard().Run() + // testmain() + NewNetTunnel().Run() } var ( diff --git a/netcard.go b/netcard.go index 25e63e1..6248acf 100644 --- a/netcard.go +++ b/netcard.go @@ -1,291 +1,138 @@ package main import ( - "bytes" - "context" - "crypto/md5" - "encoding/binary" - "encoding/gob" - "fmt" "log" "net" - "os/exec" gen "slimming/proto/gen" - "strings" "sync" - "sync/atomic" - "time" - "github.com/klauspost/compress/zstd" + "github.com/474420502/check" + "github.com/474420502/perfectshutdown" "github.com/songgao/packets/ethernet" "github.com/songgao/water" - "github.com/songgao/water/waterutil" "google.golang.org/grpc" - "google.golang.org/grpc/credentials/insecure" ) -type ExchangeBuffer struct { - BytesArray [][]byte +type NetTunnel struct { + ifce *water.Interface + shutdown *perfectshutdown.PerfectShutdown + writer chan []byte + reader chan []byte + + clients sync.Map + + gen.UnimplementedFrameServiceServer } -type NetCard struct { - FrameChan chan []byte - ifce *water.Interface - server *RPCServer - - clientMap map[string]*RPCClient - - lock sync.Mutex -} - -func (nc *NetCard) Lock(do func()) { - nc.lock.Lock() - defer nc.lock.Unlock() - do() -} - -type RPCClient struct { - FrameChan chan *ExchangeBuffer - conn *grpc.ClientConn -} - -func (cli *RPCClient) connect(realAddr string) { - log.Println("rpcclient connect", realAddr) - // defer log.Println("rpcclient exit") - - conn, err := grpc.Dial(realAddr, - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) - if err != nil { - log.Printf("did not connect: %v", err) - return +func NewNetTunnel() *NetTunnel { + var err error + nt := &NetTunnel{ + writer: make(chan []byte, 1000), + reader: make(chan []byte, 1000), + shutdown: perfectshutdown.New(), + clients: sync.Map{}, } - cli.conn = conn - go cli.run() - time.Sleep(time.Millisecond) -} - -func (cli *RPCClient) run() { - defer log.Println("rpcclient exit") - defer func() { - if err := recover(); err != nil { - log.Println("recover") - cli.conn = nil - } - }() - - cliService := gen.NewFrameServiceClient(cli.conn) - stream, err := cliService.SendFrames(context.Background()) - if err != nil { - log.Panic(err) + for ipv4, address := range config.IPv4Nodes { + nt.clients.Store(ipv4, NewRPCClient(address)) } - buf := bytes.NewBuffer(nil) - for { - // Contact the server and print out its response. - - buf.Reset() - - // encode gob - enc := gob.NewEncoder(buf) - cliBuffer := <-cli.FrameChan - - log.Printf("send to target %s source bytes: %d", waterutil.IPv4Destination(cliBuffer.BytesArray[0]), len(cliBuffer.BytesArray[0])) - err := enc.Encode(cliBuffer.BytesArray) - if err != nil { - log.Panic(err) - } - - // zstd compress - zenc, err := zstd.NewWriter(buf) - if err != nil { - log.Panic(err) - } - err = zenc.Flush() - if err != nil { - log.Println(err) - } - - // 发到对面的网卡 - err = stream.Send(&gen.Request{ - Frames: buf.Bytes(), - }) - - if err != nil { - log.Panic(err) - } - } -} - -func (nc *NetCard) Run() { - go nc.runRead() - go nc.runWrite() - // go nc.cli.run() - time.Sleep(time.Second) - - cmdstr := fmt.Sprintf("ip addr add %s dev stun", config.Network.Self.Virt) - log.Println(cmdstr) - cmd := strings.Split(cmdstr, " ") - err := exec.Command(cmd[0], cmd[1:]...).Run() - if err != nil { - log.Panic(err) - } - - cmdstr = "ip link set dev stun up" - cmd = strings.Split(cmdstr, " ") - err = exec.Command(cmd[0], cmd[1:]...).Run() - if err != nil { - log.Panic(err) - } - nc.server.run() -} - -func NewNetCard() *NetCard { - - config := water.Config{ - DeviceType: water.TUN, + // log.Println(config) + nt.ifce, err = water.New(water.Config{ + DeviceType: water.TAP, PlatformSpecificParams: water.PlatformSpecificParams{ Name: "stun", }, - } - - ifce, err := water.New(config) + }) if err != nil { log.Panic(err) } - nc := &NetCard{ - FrameChan: make(chan []byte, 3000), - ifce: ifce, - clientMap: map[string]*RPCClient{}, - } - nc.server = newRPCServer(nc) - return nc -} + go func() { -func (nc *NetCard) runRead() { - log.Println("start netcard read") + for rbuf := range nt.reader { - var ifce *water.Interface = nc.ifce + log.Println(len(rbuf)) // 数据长度 + var frame ethernet.Frame = rbuf - for { + switch frame.Ethertype() { + case ethernet.ARP: - var bytesMap map[string]*ExchangeBuffer = make(map[string]*ExchangeBuffer) + case ethernet.IPv4: - var ok bool - var isLoop int32 = 1 + } - // 20 微秒后停止收集数据 - go func() { - var after = time.NewTimer(time.Millisecond * 20) - <-after.C - atomic.StoreInt32(&isLoop, 0) - }() + // TODO: 判断地址 转入对应的客户端发送数据 - for atomic.LoadInt32(&isLoop) > 0 { - var rframe ethernet.Frame - rframe.Resize(1500) - n, err := ifce.Read([]byte(rframe)) + if nt.shutdown.IsClose() { + break + } + } + }() + + go func() { + for wbuf := range nt.writer { + var frame ethernet.Frame = wbuf + _, err := nt.ifce.Write(frame) if err != nil { - log.Fatal(err) - } - rframe = rframe[:n] - - if !waterutil.IsIPv4(rframe) || waterutil.IPv4Source(rframe).Equal(net.IPv4(0, 0, 0, 0)) { - continue + log.Println(err) } - var realAddr string = "" - config.Lock(func() { - if realAddr, ok = config.IPv4Nodes[binary.LittleEndian.Uint32([]byte(waterutil.IPv4Destination(rframe)))]; !ok { - return - } - }) - - if realAddr == "" { - log.Printf("%s is not exists", waterutil.IPv4Destination(rframe)) - continue + if nt.shutdown.IsClose() { + break } - - // log.Printf("Payload: % x\n", rframe.Payload()) - - log.Printf("Ethertype: % x %v realAddr %s\n", rframe.Ethertype(), waterutil.IsIPv4(rframe), realAddr) - log.Printf("Src %s Dst: %s\n", waterutil.IPv4Source(rframe), waterutil.IPv4Destination(rframe)) - - var buffer *ExchangeBuffer - if buffer, ok = bytesMap[realAddr]; !ok { - buffer = &ExchangeBuffer{BytesArray: make([][]byte, 0, 1000)} - bytesMap[realAddr] = buffer - } - - h := md5.New() - h.Write([]byte(rframe)) - log.Printf("%x", h.Sum(nil)) - buffer.BytesArray = append(buffer.BytesArray, []byte(rframe)) } + }() - for dst, buffer := range bytesMap { - - var cli *RPCClient - if cli, ok = nc.clientMap[dst]; !ok { - cli = &RPCClient{FrameChan: make(chan *ExchangeBuffer)} - nc.clientMap[dst] = cli - } - - if cli.conn == nil { - cli.connect(dst) - } else { - cli.FrameChan <- buffer // 网卡数据 发到对方 - } - - } - - // 写到grpc服务 - - // log.Printf("Dst: %s\n", rframe.Destination()[0:4]) - // log.Printf("Src: %s\n", rframe.Source()[0:4]) - // log.Printf("Ethertype: % x\n", rframe.Ethertype()) - // log.Printf("Payload: % x\n", rframe.Payload()) - - } + return nt } -func (nc *NetCard) runWrite() { - var ifce *water.Interface = nc.ifce - var err error +func (nt *NetTunnel) Run() { + var ( + frame ethernet.Frame + err error + n int + ) - for wframe := range nc.FrameChan { + lis, err := net.Listen("tcp", config.Network.Self.Real) + if err != nil { + log.Fatalf("failed to listen: %v", err) + } + server := grpc.NewServer() + gen.RegisterFrameServiceServer(server, nt) + log.Printf("server listening at %v", lis.Addr()) + defer server.Stop() - log.Printf("get wframes bytes len: %d", len(wframe)) - var buf = bytes.NewBuffer(wframe) + go func() { + if err := server.Serve(lis); err != nil { + log.Fatalf("failed to serve: %v", err) + } + }() - var zdec *zstd.Decoder - zdec, err = zstd.NewReader(buf) + nt.shutdown.Loop(func(index int, ps *perfectshutdown.PerfectShutdown) { + frame.Resize(1500) + n, err = nt.ifce.Read(frame) + check.CheckPanic(err) + nt.reader <- frame[0:n] // 从网卡里读数据 + }) +} + +func (nt *NetTunnel) SendFrames(stream gen.FrameService_SendFramesServer) error { + log.Printf("Start: %v", stream) + + nt.shutdown.Loop(func(index int, ps *perfectshutdown.PerfectShutdown) { + request, err := stream.Recv() if err != nil { log.Panic(err) } - zdec.Close() - - dec := gob.NewDecoder(buf) - var bufs [][]byte - err = dec.Decode(&bufs) - if err != nil { - log.Panic(err) - } - - for _, buf := range bufs { - h := md5.New() - h.Write(buf) - - log.Printf("get wframes decode len: %d, write buf hash: %x", len(buf), h.Sum(nil)) - log.Printf("get decode frames decode source: %s dst: %s", waterutil.IPv4Source(buf), waterutil.IPv4Destination(buf)) - _, err := ifce.Write(buf) - if err != nil { - log.Panic(err) + log.Printf("request: %v", len(request.Frames)) + frames := Decompress(request.GetFrames()) + if len(frames) > 0 { + for _, frame := range frames { + nt.writer <- frame } } + }) - } - + return nil } diff --git a/rpc.go b/rpc.go deleted file mode 100644 index 28e30bd..0000000 --- a/rpc.go +++ /dev/null @@ -1,56 +0,0 @@ -package main - -import ( - "log" - "net" - gen "slimming/proto/gen" - - "google.golang.org/grpc" -) - -type RPCServer struct { - gen.UnimplementedFrameServiceServer - netCard *NetCard -} - -var () - -func newRPCServer(netCard *NetCard) *RPCServer { - return &RPCServer{netCard: netCard} -} - -func (rpc *RPCServer) run() { - - lis, err := net.Listen("tcp", config.Network.Self.Real) - if err != nil { - log.Fatalf("failed to listen: %v", err) - } - s := grpc.NewServer() - gen.RegisterFrameServiceServer(s, rpc) - log.Printf("server listening at %v", lis.Addr()) - - if err := s.Serve(lis); err != nil { - log.Fatalf("failed to serve: %v", err) - } -} - -// SayHello implements helloworld.GreeterServer -func (s *RPCServer) SendFrames(stream gen.FrameService_SendFramesServer) error { - log.Printf("Start: %v", stream) - - for { - request, err := stream.Recv() - if err != nil { - log.Panic(err) - } - log.Printf("request: %v", len(request.Frames)) - s.netCard.FrameChan <- request.GetFrames() // 接受数据 广播到网卡上 - } - - // err := stream.SendAndClose(&gen.Response{Code: 0}) - // if err != nil { - // log.Panic(err) - // } - - return nil -}