slimming/netcard.go

package main

import (
	"encoding/binary"
	"fmt"
	"log"
	"net"
	"os"
	"os/exec"
	gen "slimming/proto/gen"
	"strings"
	"sync"
	"time"

	"github.com/474420502/check"
	"github.com/474420502/perfectshutdown"
	"github.com/songgao/packets/ethernet"
	"github.com/songgao/water"
	"google.golang.org/grpc"
)

type NetTunnel struct {
	ifce  *water.Interface
	ifmac net.HardwareAddr

	shutdown *perfectshutdown.PerfectShutdown
	writer   chan []byte
	reader   chan []byte

	clients sync.Map
	ipv4key any

	gen.UnimplementedFrameServiceServer
}

func NewNetTunnel() *NetTunnel {
	var err error
	nt := &NetTunnel{
		writer:   make(chan []byte, 1000),
		reader:   make(chan []byte, 1000),
		shutdown: perfectshutdown.New(),
		clients:  sync.Map{},
		ipv4key:  config.CurrentNodeKey,
	}

	nt.shutdown.OnClose(func() {
		time.Sleep(time.Second * 1)
		log.Println("OnClose")
		os.Exit(0)
	})

	for ipv4, address := range config.IPv4Nodes {
		nt.clients.Store(ipv4, NewRPCClient(address))
	}

	// log.Println(config)
	nt.ifce, err = water.New(water.Config{
		DeviceType: water.TAP,
		PlatformSpecificParams: water.PlatformSpecificParams{
			Name: "stun",
		},
	})
	if err != nil {
		log.Panic(err)
	}

	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"
	log.Println(cmdstr)
	cmd = strings.Split(cmdstr, " ")
	err = exec.Command(cmd[0], cmd[1:]...).Run()
	if err != nil {
		log.Panic(err)
	}

	ifas, err := net.Interfaces()
	if err != nil {
		log.Panic(err)
	}

	for _, ifa := range ifas {
		if ifa.Name == nt.ifce.Name() {
			nt.ifmac = ifa.HardwareAddr
			log.Println(nt.ifmac)
		}
	}

	go func() {

		for rbuf := range nt.reader {

			var frame ethernet.Frame = rbuf

			// log.Printf("% x len: %d", frame.Ethertype(), len(frame))
			switch frame.Ethertype() {
			case ethernet.ARP:
				arp := ARP(frame)
				log.Printf("bytes len: %d type(arp): % x src: %s(%s) dst: %s(%s)", len(arp), frame.Ethertype(), arp.IPv4Source(), arp.Source(), arp.IPv4Destination(), arp.Destination()) // 数据长度

				SwapBytes(frame, 0, frame, 6, 6)
				binary.BigEndian.PutUint16(frame[20:22], 2)
				SwapBytes(frame, 22, frame, 32, 10)

				nt.ifce.Write(frame)

				//TODO: 解析arp 回应arp
				// nt.clients.Range(func(key, value any) bool {
				// 	if key == nt.ipv4key {
				// 		return true
				// 	}

				// 	client := value.(*RPCClient)
				// 	client.CheckConnect()
				// 	client.Frame <- frame
				// 	log.Println(client.realAddr)
				// 	return true
				// })
			case ethernet.IPv4:
				payload := frame.Payload()

				// dst := net.IP(payload[16:20]).To4()
				// log.Printf("bytes len: %d type(ipv4): % x", len(rbuf), frame.Ethertype())
				// log.Println(frame.Source(), net.IP(payload[12:16]).To4(), binary.BigEndian.Uint16(payload[20:22]), frame.Destination(), net.IP(payload[16:20]).To4(), binary.BigEndian.Uint16(payload[22:24]))

				ipv4key := binary.BigEndian.Uint32(payload[16:20])
				if v, ok := nt.clients.Load(ipv4key); ok {
					client := v.(*RPCClient)
					client.CheckConnect()
					client.Push(frame)
					// log.Println(len(frame))
				}
			}

			// TODO: 判断地址 转入对应的客户端发送数据

			if nt.shutdown.IsClose() {
				break
			}
		}
	}()

	go func() {
		for wbuf := range nt.writer {
			var frame ethernet.Frame = wbuf
			copy(frame[0:6], nt.ifmac)

			// payload := frame.Payload()

			// log.Printf("ifce write %d  ", len(frame))
			// log.Println(frame.Source(), net.IP(payload[12:16]).To4(), binary.BigEndian.Uint16(payload[20:22]), frame.Destination(), net.IP(payload[16:20]).To4(), binary.BigEndian.Uint16(payload[22:24]))

			_, err := nt.ifce.Write(frame)
			if err != nil {
				log.Println(err)
			}

			if nt.shutdown.IsClose() {
				break
			}
		}
	}()

	return nt
}

func (nt *NetTunnel) Run() {
	var (
		frame ethernet.Frame
		err   error
		n     int
	)

	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()

	go func() {
		if err := server.Serve(lis); err != nil {
			log.Fatalf("failed to serve: %v", err)
		}
	}()

	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:  SendFrames")

	nt.shutdown.Loop(func(index int, ps *perfectshutdown.PerfectShutdown) {
		request, err := stream.Recv()
		if err != nil {
			log.Panic(err)
		}

		// buf, err := Decompress(request.GetFrames())
		// if err != nil {
		// 	log.Panic(err)
		// }
		nt.writer <- request.GetFrames()

		// frames := Decompress(request.GetFrames())
		// if len(frames) > 0 {
		// 	for _, frame := range frames {
		// 		nt.writer <- frame
		// 	}
		// }
	})

	return nil
}