yuandian/slimming
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

测试

Browse Source
This commit is contained in:
黄思敏 2022-08-31 16:28:43 +08:00
parent 84557cef46
commit 41549b64a5
6 changed files with 99 additions and 300 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
config.go
View File

@ -50,7 +50,7 @@ func NewConfig() *Config {
log.Panic(err) 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) log.Printf("real %v virt %v %v", node.Real, inet, net)
} }

6
go.mod
View File

@ -4,9 +4,13 @@ go 1.19
require github.com/songgao/water v0.0.0-20200317203138-2b4b6d7c09d8 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 ( require (
github.com/474420502/perfectshutdown v0.3.1
github.com/474420502/random v0.5.1 github.com/474420502/random v0.5.1
github.com/golang/protobuf v1.5.2 // indirect github.com/golang/protobuf v1.5.2 // indirect
github.com/songgao/packets v0.0.0-20160404182456-549a10cd4091 // indirect github.com/songgao/packets v0.0.0-20160404182456-549a10cd4091 // indirect

4
go.sum
View File

@ -1,5 +1,9 @@
cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw= 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= 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 h1:AA/r690QEKiVdCsFD9OKbjUmmtmLJQ7Qxd4IMVV1sbc=
github.com/474420502/random v0.5.1/go.mod h1:Kq5VC8ezVvzp9CUahvebIdlSiFdHzEzscaOScLvbw2c= github.com/474420502/random v0.5.1/go.mod h1:Kq5VC8ezVvzp9CUahvebIdlSiFdHzEzscaOScLvbw2c=
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=

4
main.go
View File

@ -19,8 +19,8 @@ func init() {
//go:generate bash -c "protoc --go_out=plugins=grpc:. proto/*.proto" //go:generate bash -c "protoc --go_out=plugins=grpc:. proto/*.proto"
func main() { func main() {
testmain() // testmain()
// NewNetCard().Run() NewNetTunnel().Run()
} }
var ( var (

327
netcard.go
View File

@ -1,291 +1,138 @@
package main package main
import ( import (
"bytes"
"context"
"crypto/md5"
"encoding/binary"
"encoding/gob"
"fmt"
"log" "log"
"net" "net"
"os/exec"
gen "slimming/proto/gen" gen "slimming/proto/gen"
"strings"
"sync" "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/packets/ethernet"
"github.com/songgao/water" "github.com/songgao/water"
"github.com/songgao/water/waterutil"
"google.golang.org/grpc" "google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
) )
type ExchangeBuffer struct { type NetTunnel struct {
BytesArray [][]byte ifce *water.Interface
shutdown *perfectshutdown.PerfectShutdown
writer chan []byte
reader chan []byte
clients sync.Map
gen.UnimplementedFrameServiceServer
} }
type NetCard struct { func NewNetTunnel() *NetTunnel {
FrameChan chan []byte var err error
ifce *water.Interface nt := &NetTunnel{
server *RPCServer writer: make(chan []byte, 1000),
reader: make(chan []byte, 1000),
clientMap map[string]*RPCClient shutdown: perfectshutdown.New(),
clients: sync.Map{},
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
} }
cli.conn = conn for ipv4, address := range config.IPv4Nodes {
go cli.run() nt.clients.Store(ipv4, NewRPCClient(address))
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)
} }
buf := bytes.NewBuffer(nil) // log.Println(config)
for { nt.ifce, err = water.New(water.Config{
// Contact the server and print out its response. DeviceType: water.TAP,
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,
PlatformSpecificParams: water.PlatformSpecificParams{ PlatformSpecificParams: water.PlatformSpecificParams{
Name: "stun", Name: "stun",
}, },
} })
ifce, err := water.New(config)
if err != nil { if err != nil {
log.Panic(err) log.Panic(err)
} }
nc := &NetCard{ go func() {
FrameChan: make(chan []byte, 3000),
ifce: ifce,
clientMap: map[string]*RPCClient{},
}
nc.server = newRPCServer(nc)
return nc
}
func (nc *NetCard) runRead() { for rbuf := range nt.reader {
log.Println("start netcard read")
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 微秒后停止收集数据 // TODO: 判断地址 转入对应的客户端发送数据
go func() {
var after = time.NewTimer(time.Millisecond * 20)
<-after.C
atomic.StoreInt32(&isLoop, 0)
}()
for atomic.LoadInt32(&isLoop) > 0 { if nt.shutdown.IsClose() {
var rframe ethernet.Frame break
rframe.Resize(1500) }
n, err := ifce.Read([]byte(rframe)) }
}()
go func() {
for wbuf := range nt.writer {
var frame ethernet.Frame = wbuf
_, err := nt.ifce.Write(frame)
if err != nil { if err != nil {
log.Fatal(err) log.Println(err)
}
rframe = rframe[:n]
if !waterutil.IsIPv4(rframe) || waterutil.IPv4Source(rframe).Equal(net.IPv4(0, 0, 0, 0)) {
continue
} }
var realAddr string = "" if nt.shutdown.IsClose() {
config.Lock(func() { break
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
} }
// 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 { return nt
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())
}
} }
func (nc *NetCard) runWrite() { func (nt *NetTunnel) Run() {
var ifce *water.Interface = nc.ifce var (
var err error 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)) go func() {
var buf = bytes.NewBuffer(wframe) if err := server.Serve(lis); err != nil {
log.Fatalf("failed to serve: %v", err)
}
}()
var zdec *zstd.Decoder nt.shutdown.Loop(func(index int, ps *perfectshutdown.PerfectShutdown) {
zdec, err = zstd.NewReader(buf) 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 { if err != nil {
log.Panic(err) log.Panic(err)
} }
zdec.Close() log.Printf("request: %v", len(request.Frames))
frames := Decompress(request.GetFrames())
dec := gob.NewDecoder(buf) if len(frames) > 0 {
var bufs [][]byte for _, frame := range frames {
err = dec.Decode(&bufs) nt.writer <- frame
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)
} }
} }
})
} return nil
} }

56
rpc.go
View File

@ -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
}