import (
"errors"
"net"
- "time"
)
func parseEndpoint(s string) (*net.UDPAddr, error) {
return addr, err
}
-func updateUDPConn(device *Device) error {
+func ListenerClose(l *Listener) (err error) {
+ if l.active {
+ err = CloseIPv4Socket(l.sock)
+ l.active = false
+ }
+ return
+}
+
+func (l *Listener) Init() {
+ l.update = make(chan struct{}, 1)
+ ListenerClose(l)
+}
+
+func ListeningUpdate(device *Device) error {
netc := &device.net
netc.mutex.Lock()
defer netc.mutex.Unlock()
- // close existing connection
+ // close existing sockets
- if netc.conn != nil {
- netc.conn.Close()
- netc.conn = nil
+ if err := ListenerClose(&netc.ipv4); err != nil {
+ return err
+ }
- // We need for that fd to be closed in all other go routines, which
- // means we have to wait. TODO: find less horrible way of doing this.
- time.Sleep(time.Second / 2)
+ if err := ListenerClose(&netc.ipv6); err != nil {
+ return err
}
- // open new connection
+ // open new sockets
if device.tun.isUp.Get() {
- // listen on new address
-
- conn, err := net.ListenUDP("udp", netc.addr)
- if err != nil {
- return err
+ // listen on IPv4
+
+ {
+ list := &netc.ipv6
+ sock, port, err := CreateIPv4Socket(netc.port)
+ if err != nil {
+ return err
+ }
+ netc.port = port
+ list.sock = sock
+ list.active = true
+
+ if err := SetMark(list.sock, netc.fwmark); err != nil {
+ ListenerClose(list)
+ return err
+ }
+ signalSend(list.update)
}
- // set fwmark
-
- err = SetMark(netc.conn, netc.fwmark)
- if err != nil {
- return err
+ // listen on IPv6
+
+ {
+ list := &netc.ipv6
+ sock, port, err := CreateIPv6Socket(netc.port)
+ if err != nil {
+ return err
+ }
+ netc.port = port
+ list.sock = sock
+ list.active = true
+
+ if err := SetMark(list.sock, netc.fwmark); err != nil {
+ ListenerClose(list)
+ return err
+ }
+ signalSend(list.update)
}
- // retrieve port (may have been chosen by kernel)
-
- addr := conn.LocalAddr()
- netc.conn = conn
- netc.addr, _ = net.ResolveUDPAddr(
- addr.Network(),
- addr.String(),
- )
-
- // notify goroutines
-
- signalSend(device.signal.newUDPConn)
+ // TODO: clear endpoint caches
}
return nil
}
-func closeUDPConn(device *Device) {
+func ListeningClose(device *Device) error {
netc := &device.net
netc.mutex.Lock()
- if netc.conn != nil {
- netc.conn.Close()
+ defer netc.mutex.Unlock()
+
+ if err := ListenerClose(&netc.ipv4); err != nil {
+ return err
}
- netc.mutex.Unlock()
- signalSend(device.signal.newUDPConn)
+ signalSend(netc.ipv4.update)
+
+ if err := ListenerClose(&netc.ipv6); err != nil {
+ return err
+ }
+ signalSend(netc.ipv6.update)
+
+ return nil
}
type Endpoint struct {
// source (selected based on dst type)
// (could use RawSockaddrAny and unsafe)
+ // TODO: Merge
src6 unix.RawSockaddrInet6
src4 unix.RawSockaddrInet4
src4if int32
dst unix.RawSockaddrAny
}
-type IPv4Socket int
-type IPv6Socket int
+type Socket int
+
+/* Returns a byte representation of the source field(s)
+ * for use in "under load" cookie computations.
+ */
+func (endpoint *Endpoint) Source() []byte {
+ return nil
+}
func zoneToUint32(zone string) (uint32, error) {
if zone == "" {
return uint32(n), err
}
-func CreateIPv4Socket(port int) (IPv4Socket, error) {
+func CreateIPv4Socket(port uint16) (Socket, uint16, error) {
// create socket
)
if err != nil {
- return -1, err
+ return -1, 0, err
+ }
+
+ addr := unix.SockaddrInet4{
+ Port: int(port),
}
// set sockopts and bind
if err := func() error {
-
if err := unix.SetsockoptInt(
fd,
unix.SOL_SOCKET,
return err
}
- addr := unix.SockaddrInet4{
- Port: port,
- }
return unix.Bind(fd, &addr)
-
}(); err != nil {
unix.Close(fd)
}
- return IPv4Socket(fd), err
+ return Socket(fd), uint16(addr.Port), err
}
-func CreateIPv6Socket(port int) (IPv6Socket, error) {
+func CloseIPv4Socket(sock Socket) error {
+ return unix.Close(int(sock))
+}
+
+func CloseIPv6Socket(sock Socket) error {
+ return unix.Close(int(sock))
+}
+
+func CreateIPv6Socket(port uint16) (Socket, uint16, error) {
// create socket
)
if err != nil {
- return -1, err
+ return -1, 0, err
}
// set sockopts and bind
+ addr := unix.SockaddrInet6{
+ Port: int(port),
+ }
+
if err := func() error {
if err := unix.SetsockoptInt(
return err
}
- addr := unix.SockaddrInet6{
- Port: port,
- }
return unix.Bind(fd, &addr)
}(); err != nil {
unix.Close(fd)
}
- return IPv6Socket(fd), err
+ return Socket(fd), uint16(addr.Port), err
}
func (end *Endpoint) ClearSrc() {
return errors.New("Unknown address family of source")
}
-func (end *Endpoint) ReceiveIPv4(sock IPv4Socket, buff []byte) (int, error) {
+func (end *Endpoint) ReceiveIPv4(sock Socket, buff []byte) (int, error) {
// contruct message header
return int(size), nil
}
-func (end *Endpoint) ReceiveIPv6(sock IPv6Socket, buff []byte) error {
+func (end *Endpoint) ReceiveIPv6(sock Socket, buff []byte) (int, error) {
// contruct message header
// recvmsg(sock, &mskhdr, 0)
- _, _, errno := unix.Syscall(
+ size, _, errno := unix.Syscall(
unix.SYS_RECVMSG,
uintptr(sock),
uintptr(unsafe.Pointer(&msg)),
)
if errno != 0 {
- return errno
+ return 0, errno
}
// update source cache
end.src6.Scope_id = cmsg.pktinfo.Ifindex
}
- return nil
+ return int(size), nil
}
-func SetMark(conn *net.UDPConn, value uint32) error {
- if conn == nil {
- return nil
- }
-
- file, err := conn.File()
- if err != nil {
- return err
- }
-
+func SetMark(sock Socket, value uint32) error {
return unix.SetsockoptInt(
- int(file.Fd()),
+ int(sock),
unix.SOL_SOCKET,
unix.SO_MARK,
int(value),
package main
import (
- "net"
"runtime"
"sync"
"sync/atomic"
"time"
)
+type Listener struct {
+ sock Socket
+ active bool
+ update chan struct{}
+}
+
type Device struct {
log *Logger // collection of loggers for levels
idCounter uint // for assigning debug ids to peers
}
net struct {
mutex sync.RWMutex
- addr *net.UDPAddr // UDP source address
- conn *net.UDPConn // UDP "connection"
+ ipv4 Listener
+ ipv6 Listener
+ port uint16
fwmark uint32
}
mutex sync.RWMutex
handshake chan QueueHandshakeElement
}
signal struct {
- stop chan struct{} // halts all go routines
- newUDPConn chan struct{} // a net.conn was set (consumed by the receiver routine)
+ stop chan struct{} // halts all go routines
+ updateIPv4Socket chan struct{} // a net.conn was set (consumed by the receiver routine)
+ updateIPv6Socket chan struct{} // a net.conn was set (consumed by the receiver routine)
}
underLoadUntil atomic.Value
ratelimiter Ratelimiter
device.log = NewLogger(logLevel, "("+tun.Name()+") ")
device.peers = make(map[NoisePublicKey]*Peer)
device.tun.device = tun
+
device.indices.Init()
+ device.net.ipv4.Init()
+ device.net.ipv6.Init()
device.ratelimiter.Init()
+
device.routingTable.Reset()
device.underLoadUntil.Store(time.Time{})
- // setup pools
+ // setup buffer pool
device.pool.messageBuffers = sync.Pool{
New: func() interface{} {
// prepare signals
device.signal.stop = make(chan struct{})
- device.signal.newUDPConn = make(chan struct{}, 1)
// start workers
go device.RoutineDecryption()
go device.RoutineHandshake()
}
-
+ go device.RoutineReadFromTUN()
go device.RoutineTUNEventReader()
go device.ratelimiter.RoutineGarbageCollector(device.signal.stop)
- go device.RoutineReadFromTUN()
- go device.RoutineReceiveIncomming()
-
+ go device.RoutineReceiveIncomming(&device.net.ipv4)
+ go device.RoutineReceiveIncomming(&device.net.ipv6)
return device
}
func (device *Device) Close() {
device.RemoveAllPeers()
close(device.signal.stop)
- closeUDPConn(device)
+ ListeningClose(device)
}
func (device *Device) WaitChannel() chan struct{} {
}
func main() {
+ test()
// parse arguments
)
type QueueHandshakeElement struct {
- msgType uint32
- packet []byte
- buffer *[MaxMessageSize]byte
- source *net.UDPAddr
+ msgType uint32
+ packet []byte
+ endpoint Endpoint
+ buffer *[MaxMessageSize]byte
}
type QueueInboundElement struct {
}
}
-func (device *Device) RoutineReceiveIncomming() {
+func (device *Device) RoutineReceiveIncomming(IPVersion int) {
logDebug := device.log.Debug
logDebug.Println("Routine, receive incomming, started")
+ var listener *Listener
+
+ switch IPVersion {
+ case ipv4.Version:
+ listener = &device.net.ipv4
+ case ipv6.Version:
+ listener = &device.net.ipv6
+ default:
+ return
+ }
+
for {
// wait for new conn
case <-device.signal.stop:
return
- case <-device.signal.newUDPConn:
+ case <-listener.update:
- // fetch connection
+ // fetch new socket
device.net.mutex.RLock()
- conn := device.net.conn
+ sock := listener.sock
+ okay := listener.active
device.net.mutex.RUnlock()
- if conn == nil {
+ if !okay {
continue
}
buffer := device.GetMessageBuffer()
+ var size int
+ var err error
+
for {
// read next datagram
- size, raddr, err := conn.ReadFromUDP(buffer[:])
+ var endpoint Endpoint
+
+ if IPVersion == ipv6.Version {
+ size, err = endpoint.ReceiveIPv4(sock, buffer[:])
+ } else {
+ size, err = endpoint.ReceiveIPv6(sock, buffer[:])
+ }
if err != nil {
break
buffer = device.GetMessageBuffer()
continue
- // otherwise it is a handshake related packet
+ // otherwise it is a fixed size & handshake related packet
case MessageInitiationType:
okay = len(packet) == MessageInitiationSize
device.addToHandshakeQueue(
device.queue.handshake,
QueueHandshakeElement{
- msgType: msgType,
- buffer: buffer,
- packet: packet,
- source: raddr,
+ msgType: msgType,
+ buffer: buffer,
+ packet: packet,
+ endpoint: endpoint,
},
)
buffer = device.GetMessageBuffer()
// unmarshal packet
- logDebug.Println("Process cookie reply from:", elem.source.String())
-
var reply MessageCookieReply
reader := bytes.NewReader(elem.packet)
err := binary.Read(reader, binary.LittleEndian, &reply)
projects / thirdparty / wireguard-go.git / commitdiff
