conn,device: provide 8 free bytes at packet head to conn.Bind.Send()
This enables a conn.Bind to bring its own encapsulating transport, e.g. VXLAN/Geneve. Updates tailscale/corp#27502 Signed-off-by: Jordan Whited <jordan@tailscale.com>
This commit is contained in:
parent
022546570f
commit
65cd6eed7d
10 changed files with 58 additions and 39 deletions
|
|
@ -341,7 +341,7 @@ func (e ErrUDPGSODisabled) Unwrap() error {
|
|||
return e.RetryErr
|
||||
}
|
||||
|
||||
func (s *StdNetBind) Send(bufs [][]byte, endpoint Endpoint) error {
|
||||
func (s *StdNetBind) Send(bufs [][]byte, endpoint Endpoint, offset int) error {
|
||||
s.mu.Lock()
|
||||
blackhole := s.blackhole4
|
||||
conn := s.ipv4
|
||||
|
|
@ -384,7 +384,7 @@ func (s *StdNetBind) Send(bufs [][]byte, endpoint Endpoint) error {
|
|||
)
|
||||
retry:
|
||||
if offload {
|
||||
n := coalesceMessages(ua, endpoint.(*StdNetEndpoint), bufs, *msgs, setGSOSize)
|
||||
n := coalesceMessages(ua, endpoint.(*StdNetEndpoint), bufs, offset, *msgs, setGSOSize)
|
||||
err = s.send(conn, br, (*msgs)[:n])
|
||||
if err != nil && offload && errShouldDisableUDPGSO(err) {
|
||||
offload = false
|
||||
|
|
@ -401,7 +401,7 @@ retry:
|
|||
} else {
|
||||
for i := range bufs {
|
||||
(*msgs)[i].Addr = ua
|
||||
(*msgs)[i].Buffers[0] = bufs[i]
|
||||
(*msgs)[i].Buffers[0] = bufs[i][offset:]
|
||||
setSrcControl(&(*msgs)[i].OOB, endpoint.(*StdNetEndpoint))
|
||||
}
|
||||
err = s.send(conn, br, (*msgs)[:len(bufs)])
|
||||
|
|
@ -450,7 +450,7 @@ const (
|
|||
|
||||
type setGSOFunc func(control *[]byte, gsoSize uint16)
|
||||
|
||||
func coalesceMessages(addr *net.UDPAddr, ep *StdNetEndpoint, bufs [][]byte, msgs []ipv6.Message, setGSO setGSOFunc) int {
|
||||
func coalesceMessages(addr *net.UDPAddr, ep *StdNetEndpoint, bufs [][]byte, offset int, msgs []ipv6.Message, setGSO setGSOFunc) int {
|
||||
var (
|
||||
base = -1 // index of msg we are currently coalescing into
|
||||
gsoSize int // segmentation size of msgs[base]
|
||||
|
|
@ -462,6 +462,7 @@ func coalesceMessages(addr *net.UDPAddr, ep *StdNetEndpoint, bufs [][]byte, msgs
|
|||
maxPayloadLen = maxIPv6PayloadLen
|
||||
}
|
||||
for i, buf := range bufs {
|
||||
buf = buf[offset:]
|
||||
if i > 0 {
|
||||
msgLen := len(buf)
|
||||
baseLenBefore := len(msgs[base].Buffers[0])
|
||||
|
|
|
|||
|
|
@ -98,7 +98,7 @@ func Test_coalesceMessages(t *testing.T) {
|
|||
msgs[i].Buffers = make([][]byte, 1)
|
||||
msgs[i].OOB = make([]byte, 0, 2)
|
||||
}
|
||||
got := coalesceMessages(addr, &StdNetEndpoint{AddrPort: addr.AddrPort()}, tt.buffs, msgs, mockSetGSOSize)
|
||||
got := coalesceMessages(addr, &StdNetEndpoint{AddrPort: addr.AddrPort()}, tt.buffs, 0, msgs, mockSetGSOSize)
|
||||
if got != len(tt.wantLens) {
|
||||
t.Fatalf("got len %d want: %d", got, len(tt.wantLens))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -486,7 +486,7 @@ func (bind *afWinRingBind) Send(buf []byte, nend *WinRingEndpoint, isOpen *atomi
|
|||
return winrio.SendEx(bind.rq, dataBuffer, 1, nil, addressBuffer, nil, nil, 0, 0)
|
||||
}
|
||||
|
||||
func (bind *WinRingBind) Send(bufs [][]byte, endpoint Endpoint) error {
|
||||
func (bind *WinRingBind) Send(bufs [][]byte, endpoint Endpoint, offset int) error {
|
||||
nend, ok := endpoint.(*WinRingEndpoint)
|
||||
if !ok {
|
||||
return ErrWrongEndpointType
|
||||
|
|
@ -494,6 +494,7 @@ func (bind *WinRingBind) Send(bufs [][]byte, endpoint Endpoint) error {
|
|||
bind.mu.RLock()
|
||||
defer bind.mu.RUnlock()
|
||||
for _, buf := range bufs {
|
||||
buf = buf[offset:]
|
||||
switch nend.family {
|
||||
case windows.AF_INET:
|
||||
if bind.v4.blackhole {
|
||||
|
|
|
|||
|
|
@ -107,8 +107,9 @@ func (c *ChannelBind) makeReceiveFunc(ch chan []byte) conn.ReceiveFunc {
|
|||
}
|
||||
}
|
||||
|
||||
func (c *ChannelBind) Send(bufs [][]byte, ep conn.Endpoint) error {
|
||||
func (c *ChannelBind) Send(bufs [][]byte, ep conn.Endpoint, offset int) error {
|
||||
for _, b := range bufs {
|
||||
b = b[offset:]
|
||||
select {
|
||||
case <-c.closeSignal:
|
||||
return net.ErrClosed
|
||||
|
|
|
|||
|
|
@ -45,9 +45,11 @@ type Bind interface {
|
|||
// This mark is passed to the kernel as the socket option SO_MARK.
|
||||
SetMark(mark uint32) error
|
||||
|
||||
// Send writes one or more packets in bufs to address ep. The length of
|
||||
// bufs must not exceed BatchSize().
|
||||
Send(bufs [][]byte, ep Endpoint) error
|
||||
// Send writes one or more packets in bufs to address ep. A nonzero offset
|
||||
// can be used to instruct the Bind on where packet data begins in each
|
||||
// element of the bufs slice. Space preceding offset is free to use for
|
||||
// additional encapsulation. The length of bufs must not exceed BatchSize().
|
||||
Send(bufs [][]byte, ep Endpoint, offset int) error
|
||||
|
||||
// ParseEndpoint creates a new endpoint from a string.
|
||||
ParseEndpoint(s string) (Endpoint, error)
|
||||
|
|
|
|||
|
|
@ -27,9 +27,9 @@ const (
|
|||
)
|
||||
|
||||
const (
|
||||
MinMessageSize = MessageKeepaliveSize // minimum size of transport message (keepalive)
|
||||
MaxMessageSize = MaxSegmentSize // maximum size of transport message
|
||||
MaxContentSize = MaxSegmentSize - MessageTransportSize // maximum size of transport message content
|
||||
MinMessageSize = MessageKeepaliveSize // minimum size of transport message (keepalive)
|
||||
MaxMessageSize = MaxSegmentSize // maximum size of transport message
|
||||
MaxContentSize = MaxSegmentSize - MessageTransportSize - MessageEncapsulatingTransportSize // maximum size of transport message content
|
||||
)
|
||||
|
||||
/* Implementation constants */
|
||||
|
|
|
|||
|
|
@ -426,11 +426,11 @@ type fakeBindSized struct {
|
|||
func (b *fakeBindSized) Open(port uint16) (fns []conn.ReceiveFunc, actualPort uint16, err error) {
|
||||
return nil, 0, nil
|
||||
}
|
||||
func (b *fakeBindSized) Close() error { return nil }
|
||||
func (b *fakeBindSized) SetMark(mark uint32) error { return nil }
|
||||
func (b *fakeBindSized) Send(bufs [][]byte, ep conn.Endpoint) error { return nil }
|
||||
func (b *fakeBindSized) ParseEndpoint(s string) (conn.Endpoint, error) { return nil, nil }
|
||||
func (b *fakeBindSized) BatchSize() int { return b.size }
|
||||
func (b *fakeBindSized) Close() error { return nil }
|
||||
func (b *fakeBindSized) SetMark(mark uint32) error { return nil }
|
||||
func (b *fakeBindSized) Send(bufs [][]byte, ep conn.Endpoint, offset int) error { return nil }
|
||||
func (b *fakeBindSized) ParseEndpoint(s string) (conn.Endpoint, error) { return nil, nil }
|
||||
func (b *fakeBindSized) BatchSize() int { return b.size }
|
||||
|
||||
type fakeTUNDeviceSized struct {
|
||||
size int
|
||||
|
|
|
|||
|
|
@ -61,13 +61,14 @@ const (
|
|||
)
|
||||
|
||||
const (
|
||||
MessageInitiationSize = 148 // size of handshake initiation message
|
||||
MessageResponseSize = 92 // size of response message
|
||||
MessageCookieReplySize = 64 // size of cookie reply message
|
||||
MessageTransportHeaderSize = 16 // size of data preceding content in transport message
|
||||
MessageTransportSize = MessageTransportHeaderSize + poly1305.TagSize // size of empty transport
|
||||
MessageKeepaliveSize = MessageTransportSize // size of keepalive
|
||||
MessageHandshakeSize = MessageInitiationSize // size of largest handshake related message
|
||||
MessageInitiationSize = 148 // size of handshake initiation message
|
||||
MessageResponseSize = 92 // size of response message
|
||||
MessageCookieReplySize = 64 // size of cookie reply message
|
||||
MessageTransportHeaderSize = 16 // size of data preceding content in transport message
|
||||
MessageEncapsulatingTransportSize = 8 // size of optional, free (for use by conn.Bind.Send()) space preceding the transport header
|
||||
MessageTransportSize = MessageTransportHeaderSize + poly1305.TagSize // size of empty transport
|
||||
MessageKeepaliveSize = MessageTransportSize // size of keepalive
|
||||
MessageHandshakeSize = MessageInitiationSize // size of largest handshake related message
|
||||
)
|
||||
|
||||
const (
|
||||
|
|
|
|||
|
|
@ -113,6 +113,9 @@ func (device *Device) NewPeer(pk NoisePublicKey) (*Peer, error) {
|
|||
return peer, nil
|
||||
}
|
||||
|
||||
// SendBuffers sends buffers to peer. WireGuard packet data in each element of
|
||||
// buffers must be preceded by MessageEncapsulatingTransportSize number of
|
||||
// bytes.
|
||||
func (peer *Peer) SendBuffers(buffers [][]byte) error {
|
||||
peer.device.net.RLock()
|
||||
defer peer.device.net.RUnlock()
|
||||
|
|
@ -133,7 +136,7 @@ func (peer *Peer) SendBuffers(buffers [][]byte) error {
|
|||
}
|
||||
peer.endpoint.Unlock()
|
||||
|
||||
err := peer.device.net.bind.Send(buffers, endpoint)
|
||||
err := peer.device.net.bind.Send(buffers, endpoint, MessageEncapsulatingTransportSize)
|
||||
if err == nil {
|
||||
var totalLen uint64
|
||||
for _, b := range buffers {
|
||||
|
|
|
|||
|
|
@ -45,11 +45,15 @@ import (
|
|||
*/
|
||||
|
||||
type QueueOutboundElement struct {
|
||||
buffer *[MaxMessageSize]byte // slice holding the packet data
|
||||
packet []byte // slice of "buffer" (always!)
|
||||
nonce uint64 // nonce for encryption
|
||||
keypair *Keypair // keypair for encryption
|
||||
peer *Peer // related peer
|
||||
buffer *[MaxMessageSize]byte // slice holding the packet data
|
||||
// packet is always a slice of "buffer". The starting offset in buffer
|
||||
// is either:
|
||||
// a) MessageEncapsulatingTransportSize+MessageTransportHeaderSize (plaintext)
|
||||
// b) 0 (post-encryption)
|
||||
packet []byte
|
||||
nonce uint64 // nonce for encryption
|
||||
keypair *Keypair // keypair for encryption
|
||||
peer *Peer // related peer
|
||||
}
|
||||
|
||||
type QueueOutboundElementsContainer struct {
|
||||
|
|
@ -123,14 +127,15 @@ func (peer *Peer) SendHandshakeInitiation(isRetry bool) error {
|
|||
return err
|
||||
}
|
||||
|
||||
packet := make([]byte, MessageInitiationSize)
|
||||
buf := make([]byte, MessageEncapsulatingTransportSize+MessageInitiationSize)
|
||||
packet := buf[MessageEncapsulatingTransportSize:]
|
||||
_ = msg.marshal(packet)
|
||||
peer.cookieGenerator.AddMacs(packet)
|
||||
|
||||
peer.timersAnyAuthenticatedPacketTraversal()
|
||||
peer.timersAnyAuthenticatedPacketSent()
|
||||
|
||||
err = peer.SendBuffers([][]byte{packet})
|
||||
err = peer.SendBuffers([][]byte{buf})
|
||||
if err != nil {
|
||||
peer.device.log.Errorf("%v - Failed to send handshake initiation: %v", peer, err)
|
||||
}
|
||||
|
|
@ -152,7 +157,8 @@ func (peer *Peer) SendHandshakeResponse() error {
|
|||
return err
|
||||
}
|
||||
|
||||
packet := make([]byte, MessageResponseSize)
|
||||
buf := make([]byte, MessageEncapsulatingTransportSize+MessageResponseSize)
|
||||
packet := buf[MessageEncapsulatingTransportSize:]
|
||||
_ = response.marshal(packet)
|
||||
peer.cookieGenerator.AddMacs(packet)
|
||||
|
||||
|
|
@ -167,7 +173,7 @@ func (peer *Peer) SendHandshakeResponse() error {
|
|||
peer.timersAnyAuthenticatedPacketSent()
|
||||
|
||||
// TODO: allocation could be avoided
|
||||
err = peer.SendBuffers([][]byte{packet})
|
||||
err = peer.SendBuffers([][]byte{buf})
|
||||
if err != nil {
|
||||
peer.device.log.Errorf("%v - Failed to send handshake response: %v", peer, err)
|
||||
}
|
||||
|
|
@ -184,10 +190,11 @@ func (device *Device) SendHandshakeCookie(initiatingElem *QueueHandshakeElement)
|
|||
return err
|
||||
}
|
||||
|
||||
packet := make([]byte, MessageCookieReplySize)
|
||||
buf := make([]byte, MessageEncapsulatingTransportSize+MessageCookieReplySize)
|
||||
packet := buf[MessageEncapsulatingTransportSize:]
|
||||
_ = reply.marshal(packet)
|
||||
// TODO: allocation could be avoided
|
||||
device.net.bind.Send([][]byte{packet}, initiatingElem.endpoint)
|
||||
device.net.bind.Send([][]byte{buf}, initiatingElem.endpoint, MessageEncapsulatingTransportSize)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
@ -220,7 +227,7 @@ func (device *Device) RoutineReadFromTUN() {
|
|||
elemsByPeer = make(map[*Peer]*QueueOutboundElementsContainer, batchSize)
|
||||
count = 0
|
||||
sizes = make([]int, batchSize)
|
||||
offset = MessageTransportHeaderSize
|
||||
offset = MessageEncapsulatingTransportSize + MessageTransportHeaderSize
|
||||
)
|
||||
|
||||
for i := range elems {
|
||||
|
|
@ -446,7 +453,7 @@ func (device *Device) RoutineEncryption(id int) {
|
|||
for elemsContainer := range device.queue.encryption.c {
|
||||
for _, elem := range elemsContainer.elems {
|
||||
// populate header fields
|
||||
header := elem.buffer[:MessageTransportHeaderSize]
|
||||
header := elem.buffer[MessageEncapsulatingTransportSize : MessageEncapsulatingTransportSize+MessageTransportHeaderSize]
|
||||
|
||||
fieldType := header[0:4]
|
||||
fieldReceiver := header[4:8]
|
||||
|
|
@ -469,6 +476,9 @@ func (device *Device) RoutineEncryption(id int) {
|
|||
elem.packet,
|
||||
nil,
|
||||
)
|
||||
|
||||
// re-slice packet to include encapsulating transport space
|
||||
elem.packet = elem.buffer[:MessageEncapsulatingTransportSize+len(elem.packet)]
|
||||
}
|
||||
elemsContainer.Unlock()
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue