container_discovery/internal/discoverer/ipv4/port_tcp.go

package ipv4

import (
	"encoding/json"
	"fmt"
	"net"
	"strconv"
	"sync"
	"sync/atomic"
	"time"

	"git.loafle.net/commons/logging-go"
	occu "git.loafle.net/overflow/commons-go/core/util"
	ocmd "git.loafle.net/overflow/commons-go/model/discovery"
	ocmm "git.loafle.net/overflow/commons-go/model/meta"
	"git.loafle.net/overflow/container_discovery/internal/pcap"

	"github.com/google/gopacket"
	"github.com/google/gopacket/layers"
)

func scanPortTCP(host *ocmd.Host, dp *ocmd.DiscoverPort, resultChan chan interface{}, errChan chan error, stopChan chan struct{}, wg *sync.WaitGroup) {
	defer func() {
		wg.Done()
	}()

	ps, err := pcap.RetainScanner(host.Zone)
	if nil != err {
		errChan <- fmt.Errorf("Discovery: Cannot retain pcap instance %v", err)
		return
	}
	defer func() {
		go pcap.ReleaseScanner(host.Zone)
	}()

	tcpChan := ps.OpenTCP(host.Address)
	defer func() {
		go ps.CloseTCP(host.Address, tcpChan)
	}()

	timerStopped := make(chan struct{})
	go func() {
		ports := make(map[int]*ocmd.Port)

		var delay atomic.Value
		delay.Store(false)
		ticker := time.NewTicker(time.Millisecond * 500)
		for {
			select {
			case packet, ok := <-tcpChan:
				if !ok {
					logging.Logger().Debugf("Discovery: tcp channel is closed")
					return
				}
				delay.Store(true)
				if p := handlePacketTCP(host, dp, ports, packet); nil != p {
					resultChan <- p
				}
			case <-ticker.C:
				if false == delay.Load().(bool) {
					ticker.Stop()
					timerStopped <- struct{}{}
					return
				}
				delay.Store(false)
			case <-stopChan:
				return
			}
		}
	}()

	if err := sendTCP(host, dp, stopChan); nil != err {
		errChan <- err
		return
	}

	select {
	case <-stopChan:
		return
	case <-timerStopped:
		return
	}
}

func sendTCP(host *ocmd.Host, dp *ocmd.DiscoverPort, stopChan chan struct{}) error {
	tcpPacket, err := makePacketPortTCP(host)
	if nil != err {
		return err
	}
	defer func() {
		tcpPacket.PacketConn.Close()
	}()

	buf := gopacket.NewSerializeBuffer()

Loop:
	for portNumber := dp.FirstScanRange; portNumber < dp.LastScanRange; portNumber++ {
		if nil != dp.ExcludePorts {
			for _, exPortNumber := range dp.ExcludePorts {
				if portNumber == exPortNumber {
					continue Loop
				}
			}
		}
		tcpPacket.TCP.DstPort = layers.TCPPort(portNumber)
		tcpPacket.TCP.SetNetworkLayerForChecksum(tcpPacket.IP)

		if err := gopacket.SerializeLayers(buf, tcpPacket.Opts, tcpPacket.TCP); err != nil {
			return err
		}
		if _, err := tcpPacket.PacketConn.WriteTo(buf.Bytes(), &net.IPAddr{IP: tcpPacket.IP.DstIP}); err != nil {
			return err
		}

		timer := time.NewTimer(time.Microsecond * 100)

		select {
		case <-stopChan:
			return nil
		case <-timer.C:
		}
	}

	return nil
}

func handlePacketTCP(host *ocmd.Host, dp *ocmd.DiscoverPort, ports map[int]*ocmd.Port, packet *layers.TCP) *ocmd.Port {
	if nil == packet || packet.DstPort != 60000 {
		return nil
	}

	if packet.RST {
		return nil
	}

	port := int(packet.SrcPort)
	logging.Logger().Debugf("Discovery: IP of TCP(%d) src %s", port, host.Address)

	if _, ok := ports[port]; ok || !dp.Contains(port) {
		return nil
	}

	p := &ocmd.Port{
		MetaPortType:   ocmm.ToMetaPortType(ocmm.MetaPortTypeEnumTCP),
		PortNumber:     json.Number(strconv.Itoa(port)),
		DiscoveredDate: occu.NowPtr(),
	}
	p.Host = host

	ports[port] = p

	return p
}

type PortPacketTCP struct {
	IP         *layers.IPv4
	TCP        *layers.TCP
	Opts       gopacket.SerializeOptions
	PacketConn net.PacketConn
}

func makePacketPortTCP(host *ocmd.Host) (*PortPacketTCP, error) {
	packetTCP := &PortPacketTCP{}

	srcIP := net.ParseIP(host.Zone.Address)
	if nil == srcIP {
		return nil, fmt.Errorf("Discovery: IP(%s) of zone is not valid", host.Zone.Address)
	}
	dstIP := net.ParseIP(host.Address)
	if nil == dstIP {
		return nil, fmt.Errorf("Discovery: IP(%s) of host is not valid", host.Address)
	}

	packetTCP.IP = &layers.IPv4{
		SrcIP:    srcIP.To4(),
		DstIP:    dstIP.To4(),
		Version:  4,
		TTL:      64,
		Protocol: layers.IPProtocolTCP,
	}
	packetTCP.TCP = &layers.TCP{
		SrcPort: 60000,
		DstPort: 0, // will be incremented during the scan
		SYN:     true,
		Seq:     0,
	}
	packetTCP.Opts = gopacket.SerializeOptions{
		ComputeChecksums: true,
		FixLengths:       true,
	}

	conn, err := net.ListenPacket("ip4:tcp", "0.0.0.0")
	if err != nil {
		return nil, fmt.Errorf("Discovery: SYN create socket error %v", err)
	}

	packetTCP.PacketConn = conn

	return packetTCP, nil
}