summaryrefslogtreecommitdiffhomepage
path: root/src/peer.go
blob: f3eb6c28518e74377715d259c04c4624103695b5 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
package main

import (
	"encoding/base64"
	"errors"
	"fmt"
	"sync"
	"time"
)

type Peer struct {
	id                          uint
	mutex                       sync.RWMutex
	persistentKeepaliveInterval uint64
	keyPairs                    KeyPairs
	handshake                   Handshake
	device                      *Device
	endpoint                    Endpoint
	stats                       struct {
		txBytes           uint64 // bytes send to peer (endpoint)
		rxBytes           uint64 // bytes received from peer
		lastHandshakeNano int64  // nano seconds since epoch
	}
	time struct {
		mutex         sync.RWMutex
		lastSend      time.Time // last send message
		lastHandshake time.Time // last completed handshake
		nextKeepalive time.Time
	}
	signal struct {
		newKeyPair         chan struct{} // (size 1) : a new key pair was generated
		handshakeBegin     chan struct{} // (size 1) : request that a new handshake be started ("queue handshake")
		handshakeCompleted chan struct{} // (size 1) : handshake completed
		handshakeReset     chan struct{} // (size 1) : reset handshake negotiation state
		flushNonceQueue    chan struct{} // (size 1) : empty queued packets
		messageSend        chan struct{} // (size 1) : a message was send to the peer
		messageReceived    chan struct{} // (size 1) : an authenticated message was received
		stop               chan struct{} // (size 0) : close to stop all goroutines for peer
	}
	timer struct {
		// state related to WireGuard timers

		keepalivePersistent *time.Timer // set for persistent keepalives
		keepalivePassive    *time.Timer // set upon recieving messages
		newHandshake        *time.Timer // begin a new handshake (after Keepalive + RekeyTimeout)
		zeroAllKeys         *time.Timer // zero all key material (after RejectAfterTime*3)
		handshakeDeadline   *time.Timer // Current handshake must be completed

		pendingKeepalivePassive bool
		pendingNewHandshake     bool
		pendingZeroAllKeys      bool

		needAnotherKeepalive    bool
		sendLastMinuteHandshake bool
	}
	queue struct {
		nonce    chan *QueueOutboundElement // nonce / pre-handshake queue
		outbound chan *QueueOutboundElement // sequential ordering of work
		inbound  chan *QueueInboundElement  // sequential ordering of work
	}
	mac CookieGenerator
}

func (device *Device) NewPeer(pk NoisePublicKey) (*Peer, error) {
	// create peer

	peer := new(Peer)
	peer.mutex.Lock()
	defer peer.mutex.Unlock()

	peer.mac.Init(pk)
	peer.device = device

	peer.timer.keepalivePersistent = NewStoppedTimer()
	peer.timer.keepalivePassive = NewStoppedTimer()
	peer.timer.newHandshake = NewStoppedTimer()
	peer.timer.zeroAllKeys = NewStoppedTimer()

	// assign id for debugging

	device.mutex.Lock()
	peer.id = device.idCounter
	device.idCounter += 1

	// check if over limit

	if len(device.peers) >= MaxPeers {
		return nil, errors.New("Too many peers")
	}

	// map public key

	_, ok := device.peers[pk]
	if ok {
		return nil, errors.New("Adding existing peer")
	}
	device.peers[pk] = peer
	device.mutex.Unlock()

	// precompute DH

	handshake := &peer.handshake
	handshake.mutex.Lock()
	handshake.remoteStatic = pk
	handshake.precomputedStaticStatic = device.privateKey.sharedSecret(handshake.remoteStatic)
	handshake.mutex.Unlock()

	// reset endpoint

	peer.endpoint = nil

	// prepare queuing

	peer.queue.nonce = make(chan *QueueOutboundElement, QueueOutboundSize)
	peer.queue.outbound = make(chan *QueueOutboundElement, QueueOutboundSize)
	peer.queue.inbound = make(chan *QueueInboundElement, QueueInboundSize)

	// prepare signaling & routines

	peer.signal.stop = make(chan struct{})
	peer.signal.newKeyPair = make(chan struct{}, 1)
	peer.signal.handshakeBegin = make(chan struct{}, 1)
	peer.signal.handshakeReset = make(chan struct{}, 1)
	peer.signal.handshakeCompleted = make(chan struct{}, 1)
	peer.signal.flushNonceQueue = make(chan struct{}, 1)

	go peer.RoutineNonce()
	go peer.RoutineTimerHandler()
	go peer.RoutineHandshakeInitiator()
	go peer.RoutineSequentialSender()
	go peer.RoutineSequentialReceiver()

	return peer, nil
}

func (peer *Peer) SendBuffer(buffer []byte) error {
	peer.device.net.mutex.RLock()
	defer peer.device.net.mutex.RUnlock()
	peer.mutex.RLock()
	defer peer.mutex.RUnlock()
	if peer.endpoint == nil {
		return errors.New("No known endpoint for peer")
	}
	return peer.device.net.bind.Send(buffer, peer.endpoint)
}

/* Returns a short string identification for logging
 */
func (peer *Peer) String() string {
	if peer.endpoint == nil {
		return fmt.Sprintf(
			"peer(%d unknown %s)",
			peer.id,
			base64.StdEncoding.EncodeToString(peer.handshake.remoteStatic[:]),
		)
	}
	return fmt.Sprintf(
		"peer(%d %s %s)",
		peer.id,
		peer.endpoint.DstToString(),
		base64.StdEncoding.EncodeToString(peer.handshake.remoteStatic[:]),
	)
}

func (peer *Peer) Close() {
	close(peer.signal.stop)
}