解析tcp segment 报文与tcp配置

tcpip/header/tcp.go

@@ -4,6 +4,7 @@ import (
 	"encoding/binary"
 	"fmt"
 	"netstack/tcpip"
+	"netstack/tcpip/seqnum"
 )
 
 /*
@@ -39,6 +40,10 @@ type TCPFields struct {
 	DstPort uint16
 
 	// SeqNum is the "sequence number" field of a TCP packet.
+	// TCP的初始序列号ISN是随机生成的
+	// 如果TCP每次连接都使用固定ISN，黑客可以很方便模拟任何IP与server建立连接
+	// 如果ISN是固定的，那很可能在新连接建立后，上次连接通信的报文才到达，
+	// 这种情况有概率发生老报文的seq号正好是server希望收到的新连接的报文seq。这就全乱了。
 	SeqNum uint32
 
 	// AckNum is the "acknowledgement number" field of a TCP packet.
@@ -72,6 +77,42 @@ const (
 	urgentPtr   = 18
 )
 
+// Options that may be present in a TCP segment.
+const (
+	TCPOptionEOL           = 0
+	TCPOptionNOP           = 1
+	TCPOptionMSS           = 2
+	TCPOptionWS            = 3
+	TCPOptionTS            = 8
+	TCPOptionSACKPermitted = 4
+	TCPOptionSACK          = 5
+)
+
+// SACKBlock 表示 sack 块的结构体
+type SACKBlock struct {
+	// Start indicates the lowest sequence number in the block.
+	Start seqnum.Value
+
+	// End indicates the sequence number immediately following the last
+	// sequence number of this block.
+	End seqnum.Value
+}
+
+// TCPOptions tcp选项结构，这个结构不表示 syn/syn-ack 报文
+type TCPOptions struct {
+	// TS is true if the TimeStamp option is enabled.
+	TS bool
+
+	// TSVal is the value in the TSVal field of the segment.
+	TSVal uint32
+
+	// TSEcr is the value in the TSEcr field of the segment.
+	TSEcr uint32
+
+	// SACKBlocks are the SACK blocks specified in the segment.
+	SACKBlocks []SACKBlock
+}
+
 // TCP represents a TCP header stored in a byte array.
 type TCP []byte
 
@@ -163,6 +204,51 @@ func (b TCP) Options() []byte {
 	return b[TCPMinimumSize:b.DataOffset()]
 }
 
+// ParseTCPOptions extracts and stores all known options in the provided byte
+// slice in a TCPOptions structure.
+func ParseTCPOptions(b []byte) TCPOptions {
+	opts := TCPOptions{}
+	limit := len(b)
+	for i := 0; i < limit; {
+		switch b[i] {
+		case TCPOptionEOL: // 末尾
+			i = limit
+		case TCPOptionNOP: // 空值
+			i++
+		case TCPOptionTS: // 计时
+			if i+10 > limit || (b[i+1] != 10) {
+				return opts
+			}
+			opts.TS = true
+			opts.TSVal = binary.BigEndian.Uint32(b[i+2:])
+			opts.TSEcr = binary.BigEndian.Uint32(b[i+6:])
+			i += 10
+		case TCPOptionSACK:
+			if i+2 > limit {
+				// Malformed SACK block, just return and stop parsing.
+				return opts
+			}
+			sackOptionLen := int(b[i+1])
+			// TODO 需要添加
+
+			i += sackOptionLen
+		default:
+			// We don't recognize this option, just skip over it.
+			if i+2 > limit {
+				return opts
+			}
+			l := int(b[i+1])
+			// If the length is incorrect or if l+i overflows the
+			// total options length then return false.
+			if l < 2 || i+l > limit {
+				return opts
+			}
+			i++
+		}
+	}
+	return opts
+}
+
 /*
  0                   1                   2                   3
  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
@@ -191,9 +277,9 @@ var tcpFmt string = `
 |% 32s |
 |% 4s|% 4s|%06b|% 16s|
 |% 16s|% 16s|
-|      Options      |   Padding   |
+|% 8v|
-%v
+|             Padding             |
-`
+%v`
 
 func (b TCP) String() string {
 	return fmt.Sprintf(tcpFmt, atoi(b.SourcePort()), atoi(b.DestinationPort()),
@@ -201,5 +287,6 @@ func (b TCP) String() string {
 		atoi(b.AckNumber()),
 		atoi(b.DataOffset()), "0", b.Flags(), atoi(b.WindowSize()),
 		atoi(b.Checksum()), atoi(b.UrgentPtr()),
+		ParseTCPOptions(b.Options()),
 		b.viewPayload())
 }

tcpip/transport/tcp/endpoint.go

@@ -1,7 +1,9 @@
 package tcp
 
 import (
+	"fmt"
 	"log"
+	"netstack/sleep"
 	"netstack/tcpip"
 	"netstack/tcpip/buffer"
 	"netstack/tcpip/header"
@@ -34,6 +36,13 @@ type endpoint struct {
 
 	// TODO 需要添加
 
+	// rcvListMu can be taken after the endpoint mu below.
+	rcvListMu  sync.Mutex
+	rcvList    segmentList
+	rcvClosed  bool
+	rcvBufSize int
+	rcvBufUsed int
+
 	// The following fields are protected by the mutex.
 	mu                sync.RWMutex
 	id                stack.TransportEndpointID // tcp端在网络协议栈的唯一ID
@@ -56,6 +65,19 @@ type endpoint struct {
 	// workerRunning specifies if a worker goroutine is running.
 	workerRunning bool
 
+	segmentQueue segmentQueue
+
+	// When the send side is closed, the protocol goroutine is notified via
+	// sndCloseWaker, and sndClosed is set to true.
+	sndBufMu      sync.Mutex
+	sndBufSize    int
+	sndBufUsed    int
+	sndClosed     bool
+	sndBufInQueue seqnum.Size
+	sndQueue      segmentList
+	sndWaker      sleep.Waker
+	sndCloseWaker sleep.Waker
+
 	// acceptedChan is used by a listening endpoint protocol goroutine to
 	// send newly accepted connections to the endpoint so that they can be
 	// read by Accept() calls.
@@ -71,9 +93,12 @@ func newEndpoint(stack *stack.Stack, netProto tcpip.NetworkProtocolNumber, waite
 		stack:       stack,
 		netProto:    netProto,
 		waiterQueue: waiterQueue,
+		rcvBufSize:  DefaultBufferSize,
+		sndBufSize:  DefaultBufferSize,
 	}
 	// TODO 需要添加
 	log.Println("新建tcp端")
+	e.segmentQueue.setLimit(2 * e.rcvBufSize)
 	return e
 }
 
@@ -296,7 +321,6 @@ func (e *endpoint) GetSockOpt(opt interface{}) *tcpip.Error {
 }
 
 func (e *endpoint) HandlePacket(r *stack.Route, id stack.TransportEndpointID, vv buffer.VectorisedView) {
-	log.Println("接收到数据")
 	s := newSegment(r, id, vv)
 	// 解析tcp段，如果解析失败，丢弃该报文
 	if !s.parse() {
@@ -307,7 +331,20 @@ func (e *endpoint) HandlePacket(r *stack.Route, id stack.TransportEndpointID, vv
 	}
 
 	e.stack.Stats().TCP.ValidSegmentsReceived.Increment() // 有效报文喜加一
-	log.Println(s)
+	if (s.flags & flagRst) != 0 {                         // RST报文需要拒绝
+		e.stack.Stats().TCP.ResetsReceived.Increment()
+	}
+	// Send packet to worker goroutine.
+	if e.segmentQueue.enqueue(s) {
+		log.Printf("收到 tcp [%s] 报文片段 from %s, seq: %d, ack: %d",
+			flagString(s.flags), fmt.Sprintf("%s:%d", s.id.RemoteAddress, s.id.RemotePort),
+			s.sequenceNumber, s.ackNumber)
+		//e.newSegmentWaker.Assert()
+	} else {
+		// The queue is full, so we drop the segment.
+		e.stack.Stats().DroppedPackets.Increment()
+		s.decRef()
+	}
 }
 
 func (e *endpoint) HandleControlPacket(id stack.TransportEndpointID, typ stack.ControlType, extra uint32, vv buffer.VectorisedView) {

tcpip/transport/tcp/segment.go

@@ -4,11 +4,46 @@ import (
 	"log"
 	"netstack/tcpip/buffer"
 	"netstack/tcpip/header"
+	"netstack/tcpip/seqnum"
 	"netstack/tcpip/stack"
+	"strings"
 	"sync/atomic"
 )
 
-// tcp 太复杂了 专门写一个协议解析器
+// tcp 太复杂了 专门写一个协议解析器 segment 是有种类之分的
+
+// Flags that may be set in a TCP segment.
+const (
+	flagFin = 1 << iota
+	flagSyn
+	flagRst
+	flagPsh
+	flagAck
+	flagUrg
+)
+
+func flagString(flags uint8) string {
+	var s []string
+	if (flags & flagAck) != 0 {
+		s = append(s, "ack")
+	}
+	if (flags & flagFin) != 0 {
+		s = append(s, "fin")
+	}
+	if (flags & flagPsh) != 0 {
+		s = append(s, "psh")
+	}
+	if (flags & flagRst) != 0 {
+		s = append(s, "rst")
+	}
+	if (flags & flagSyn) != 0 {
+		s = append(s, "syn")
+	}
+	if (flags & flagUrg) != 0 {
+		s = append(s, "urg")
+	}
+	return strings.Join(s, "|")
+}
 
 // segment 表示一个 TCP 段。它保存有效负载和解析的 TCP 段信息，并且可以添加到侵入列表中
 type segment struct {
@@ -20,7 +55,15 @@ type segment struct {
 	// views is used as buffer for data when its length is large
 	// enough to store a VectorisedView.
 	views [8]buffer.View
-	// TODO 需要添加
+	// TODO 需要解析
+	viewToDeliver  int
+	sequenceNumber seqnum.Value // tcp序号 第一个字节在整个报文的位置
+	ackNumber      seqnum.Value // 确认号 希望继续获取的下一个字节序号
+	flags          uint8
+	window         seqnum.Size
+	// parsedOptions stores the parsed values from the options in the segment.
+	parsedOptions header.TCPOptions
+	options       []byte
 }
 
 func newSegment(r *stack.Route, id stack.TransportEndpointID, vv buffer.VectorisedView) *segment {
@@ -41,5 +84,18 @@ func (s *segment) incRef() {
 
 func (s *segment) parse() bool {
 	log.Println(header.TCP(s.data.First()))
+	h := header.TCP(s.data.First())
+	offset := int(h.DataOffset())
+	if offset < header.TCPMinimumSize || offset > len(h) {
 		return false
 	}
+	s.options = h.Options()
+	//s.parsedOptions = header.ParseTCPOptions(s.options)
+	s.data.TrimFront(offset)
+
+	s.sequenceNumber = seqnum.Value(h.SequenceNumber())
+	s.ackNumber = seqnum.Value(h.AckNumber())
+	s.flags = h.Flags() // U|A|P|R|S|F
+	s.window = seqnum.Size(h.WindowSize())
+	return true
+}

tcpip/transport/tcp/segment_queue.go

@@ -0,0 +1,50 @@
+package tcp
+
+import (
+	"netstack/tcpip/header"
+	"sync"
+)
+
+type segmentQueue struct {
+	mu    sync.Mutex
+	list  segmentList // 队列实现
+	limit int         // 队列容量
+	used  int         // 队列长度
+}
+
+func (q *segmentQueue) empty() bool {
+	q.mu.Lock()
+	r := q.used == 0
+	q.mu.Unlock()
+	return r
+}
+
+func (q *segmentQueue) enqueue(s *segment) bool {
+	q.mu.Lock()
+	r := q.used < q.limit
+	if r {
+		q.list.PushBack(s)
+		q.used += s.data.Size() + header.TCPMinimumSize
+	}
+	q.mu.Unlock()
+
+	return r
+}
+
+func (q *segmentQueue) dequeue() *segment {
+	q.mu.Lock()
+	s := q.list.Front()
+	if s != nil {
+		q.list.Remove(s)
+		q.used -= s.data.Size() + header.TCPMinimumSize
+	}
+	q.mu.Unlock()
+
+	return s
+}
+
+func (q *segmentQueue) setLimit(limit int) {
+	q.mu.Lock()
+	q.limit = limit
+	q.mu.Unlock()
+}