forked from Jigsaw-Code/outline-ss-server
-
Notifications
You must be signed in to change notification settings - Fork 0
/
tcp.go
408 lines (363 loc) · 14.9 KB
/
tcp.go
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
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
// Copyright 2018 Jigsaw Operations LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package service
import (
"bytes"
"container/list"
"context"
"errors"
"fmt"
"io"
"net"
"sync"
"syscall"
"time"
"github.com/Jigsaw-Code/outline-sdk/transport"
"github.com/Jigsaw-Code/outline-sdk/transport/shadowsocks"
"github.com/OneLC/outline-ss-server/ipinfo"
onet "github.com/OneLC/outline-ss-server/net"
"github.com/OneLC/outline-ss-server/service/metrics"
logging "github.com/op/go-logging"
"github.com/shadowsocks/go-shadowsocks2/socks"
)
// TCPMetrics is used to report metrics on TCP connections.
type TCPMetrics interface {
ipinfo.IPInfoMap
// TCP metrics
AddOpenTCPConnection(clientInfo ipinfo.IPInfo)
AddAuthenticatedTCPConnection(clientAddr net.Addr, accessKey string)
AddClosedTCPConnection(clientInfo ipinfo.IPInfo, clientAddr net.Addr, accessKey string, status string, data metrics.ProxyMetrics, duration time.Duration)
AddTCPProbe(status, drainResult string, port int, clientProxyBytes int64)
}
func remoteIP(conn net.Conn) net.IP {
addr := conn.RemoteAddr()
if addr == nil {
return nil
}
if tcpaddr, ok := addr.(*net.TCPAddr); ok {
return tcpaddr.IP
}
ipstr, _, err := net.SplitHostPort(addr.String())
if err == nil {
return net.ParseIP(ipstr)
}
return nil
}
// Wrapper for logger.Debugf during TCP access key searches.
func debugTCP(cipherID, template string, val interface{}) {
// This is an optimization to reduce unnecessary allocations due to an interaction
// between Go's inlining/escape analysis and varargs functions like logger.Debugf.
if logger.IsEnabledFor(logging.DEBUG) {
logger.Debugf("TCP(%s): "+template, cipherID, val)
}
}
// bytesForKeyFinding is the number of bytes to read for finding the AccessKey.
// Is must satisfy provided >= bytesForKeyFinding >= required for every cipher in the list.
// provided = saltSize + 2 + 2 * cipher.TagSize, the minimum number of bytes we will see in a valid connection
// required = saltSize + 2 + cipher.TagSize, the number of bytes needed to authenticate the connection.
const bytesForKeyFinding = 50
func findAccessKey(clientReader io.Reader, clientIP net.IP, cipherList CipherList) (*CipherEntry, io.Reader, []byte, time.Duration, error) {
// We snapshot the list because it may be modified while we use it.
ciphers := cipherList.SnapshotForClientIP(clientIP)
firstBytes := make([]byte, bytesForKeyFinding)
if n, err := io.ReadFull(clientReader, firstBytes); err != nil {
return nil, clientReader, nil, 0, fmt.Errorf("reading header failed after %d bytes: %w", n, err)
}
findStartTime := time.Now()
entry, elt := findEntry(firstBytes, ciphers)
timeToCipher := time.Since(findStartTime)
if entry == nil {
// TODO: Ban and log client IPs with too many failures too quick to protect against DoS.
return nil, clientReader, nil, timeToCipher, fmt.Errorf("could not find valid TCP cipher")
}
// Move the active cipher to the front, so that the search is quicker next time.
cipherList.MarkUsedByClientIP(elt, clientIP)
salt := firstBytes[:entry.CryptoKey.SaltSize()]
return entry, io.MultiReader(bytes.NewReader(firstBytes), clientReader), salt, timeToCipher, nil
}
// Implements a trial decryption search. This assumes that all ciphers are AEAD.
func findEntry(firstBytes []byte, ciphers []*list.Element) (*CipherEntry, *list.Element) {
// To hold the decrypted chunk length.
chunkLenBuf := [2]byte{}
for ci, elt := range ciphers {
entry := elt.Value.(*CipherEntry)
cryptoKey := entry.CryptoKey
_, err := shadowsocks.Unpack(chunkLenBuf[:0], firstBytes[:cryptoKey.SaltSize()+2+cryptoKey.TagSize()], cryptoKey)
if err != nil {
debugTCP(entry.ID, "Failed to decrypt length: %v", err)
continue
}
debugTCP(entry.ID, "Found cipher at index %d", ci)
return entry, elt
}
return nil, nil
}
type StreamAuthenticateFunc func(clientConn transport.StreamConn) (string, transport.StreamConn, *onet.ConnectionError)
// ShadowsocksTCPMetrics is used to report Shadowsocks metrics on TCP connections.
type ShadowsocksTCPMetrics interface {
// Shadowsocks TCP metrics
AddTCPCipherSearch(accessKeyFound bool, timeToCipher time.Duration)
}
// NewShadowsocksStreamAuthenticator creates a stream authenticator that uses Shadowsocks.
// TODO(fortuna): Offer alternative transports.
func NewShadowsocksStreamAuthenticator(ciphers CipherList, replayCache *ReplayCache, metrics ShadowsocksTCPMetrics) StreamAuthenticateFunc {
return func(clientConn transport.StreamConn) (string, transport.StreamConn, *onet.ConnectionError) {
// Find the cipher and acess key id.
cipherEntry, clientReader, clientSalt, timeToCipher, keyErr := findAccessKey(clientConn, remoteIP(clientConn), ciphers)
metrics.AddTCPCipherSearch(keyErr == nil, timeToCipher)
if keyErr != nil {
const status = "ERR_CIPHER"
return "", nil, onet.NewConnectionError(status, "Failed to find a valid cipher", keyErr)
}
var id string
if cipherEntry != nil {
id = cipherEntry.ID
}
// Check if the connection is a replay.
isServerSalt := cipherEntry.SaltGenerator.IsServerSalt(clientSalt)
// Only check the cache if findAccessKey succeeded and the salt is unrecognized.
if isServerSalt || !replayCache.Add(cipherEntry.ID, clientSalt) {
var status string
if isServerSalt {
status = "ERR_REPLAY_SERVER"
} else {
status = "ERR_REPLAY_CLIENT"
}
return id, nil, onet.NewConnectionError(status, "Replay detected", nil)
}
ssr := shadowsocks.NewReader(clientReader, cipherEntry.CryptoKey)
ssw := shadowsocks.NewWriter(clientConn, cipherEntry.CryptoKey)
ssw.SetSaltGenerator(cipherEntry.SaltGenerator)
return id, transport.WrapConn(clientConn, ssr, ssw), nil
}
}
type tcpHandler struct {
port int
m TCPMetrics
readTimeout time.Duration
authenticate StreamAuthenticateFunc
dialer transport.StreamDialer
}
// NewTCPService creates a TCPService
func NewTCPHandler(port int, authenticate StreamAuthenticateFunc, m TCPMetrics, timeout time.Duration) TCPHandler {
return &tcpHandler{
port: port,
m: m,
readTimeout: timeout,
authenticate: authenticate,
dialer: defaultDialer,
}
}
var defaultDialer = makeValidatingTCPStreamDialer(onet.RequirePublicIP)
func makeValidatingTCPStreamDialer(targetIPValidator onet.TargetIPValidator) transport.StreamDialer {
return &transport.TCPDialer{Dialer: net.Dialer{Control: func(network, address string, c syscall.RawConn) error {
ip, _, _ := net.SplitHostPort(address)
return targetIPValidator(net.ParseIP(ip))
}}}
}
// TCPService is a Shadowsocks TCP service that can be started and stopped.
type TCPHandler interface {
Handle(ctx context.Context, conn transport.StreamConn)
// SetTargetDialer sets the [transport.StreamDialer] to be used to connect to target addresses.
SetTargetDialer(dialer transport.StreamDialer)
}
func (s *tcpHandler) SetTargetDialer(dialer transport.StreamDialer) {
s.dialer = dialer
}
func ensureConnectionError(err error, fallbackStatus string, fallbackMsg string) *onet.ConnectionError {
if err == nil {
return nil
}
var connErr *onet.ConnectionError
if errors.As(err, &connErr) {
return connErr
} else {
return onet.NewConnectionError(fallbackStatus, fallbackMsg, err)
}
}
type StreamListener func() (transport.StreamConn, error)
func WrapStreamListener[T transport.StreamConn](f func() (T, error)) StreamListener {
return func() (transport.StreamConn, error) {
return f()
}
}
type StreamHandler func(ctx context.Context, conn transport.StreamConn)
// StreamServe repeatedly calls `accept` to obtain connections and `handle` to handle them until
// accept() returns [ErrClosed]. When that happens, all connection handlers will be notified
// via their [context.Context]. StreamServe will return after all pending handlers return.
func StreamServe(accept StreamListener, handle StreamHandler) {
var running sync.WaitGroup
defer running.Wait()
ctx, contextCancel := context.WithCancel(context.Background())
defer contextCancel()
for {
clientConn, err := accept()
if err != nil {
if errors.Is(err, net.ErrClosed) {
break
}
logger.Warningf("AcceptTCP failed: %v. Continuing to listen.", err)
continue
}
running.Add(1)
go func() {
defer running.Done()
defer clientConn.Close()
defer func() {
if r := recover(); r != nil {
logger.Warningf("Panic in TCP handler: %v. Continuing to listen.", r)
}
}()
handle(ctx, clientConn)
}()
}
}
func (h *tcpHandler) Handle(ctx context.Context, clientConn transport.StreamConn) {
clientInfo, err := ipinfo.GetIPInfoFromAddr(h.m, clientConn.RemoteAddr())
if err != nil {
logger.Warningf("Failed client info lookup: %v", err)
}
logger.Debugf("Got info \"%#v\" for IP %v", clientInfo, clientConn.RemoteAddr().String())
h.m.AddOpenTCPConnection(clientInfo)
var proxyMetrics metrics.ProxyMetrics
measuredClientConn := metrics.MeasureConn(clientConn, &proxyMetrics.ProxyClient, &proxyMetrics.ClientProxy)
connStart := time.Now()
id, connError := h.handleConnection(ctx, h.port, clientInfo, measuredClientConn, &proxyMetrics)
connDuration := time.Since(connStart)
status := "OK"
if connError != nil {
status = connError.Status
logger.Debugf("TCP Error: %v: %v", connError.Message, connError.Cause)
}
h.m.AddClosedTCPConnection(clientInfo, clientConn.RemoteAddr(), id, status, proxyMetrics, connDuration)
measuredClientConn.Close() // Closing after the metrics are added aids integration testing.
logger.Debugf("Done with status %v, duration %v", status, connDuration)
}
func getProxyRequest(clientConn transport.StreamConn) (string, error) {
// TODO(fortuna): Use Shadowsocks proxy, HTTP CONNECT or SOCKS5 based on first byte:
// case 1, 3 or 4: Shadowsocks (address type)
// case 5: SOCKS5 (protocol version)
// case "C": HTTP CONNECT (first char of method)
tgtAddr, err := socks.ReadAddr(clientConn)
if err != nil {
return "", err
}
return tgtAddr.String(), nil
}
func proxyConnection(ctx context.Context, dialer transport.StreamDialer, tgtAddr string, clientConn transport.StreamConn) *onet.ConnectionError {
tgtConn, dialErr := dialer.DialStream(ctx, tgtAddr)
if dialErr != nil {
// We don't drain so dial errors and invalid addresses are communicated quickly.
return ensureConnectionError(dialErr, "ERR_CONNECT", "Failed to connect to target")
}
defer tgtConn.Close()
logger.Debugf("proxy %s <-> %s", clientConn.RemoteAddr().String(), tgtConn.RemoteAddr().String())
fromClientErrCh := make(chan error)
go func() {
_, fromClientErr := io.Copy(tgtConn, clientConn)
if fromClientErr != nil {
// Drain to prevent a close in the case of a cipher error.
io.Copy(io.Discard, clientConn)
}
clientConn.CloseRead()
// Send FIN to target.
// We must do this after the drain is completed, otherwise the target will close its
// connection with the proxy, which will, in turn, close the connection with the client.
tgtConn.CloseWrite()
fromClientErrCh <- fromClientErr
}()
_, fromTargetErr := io.Copy(clientConn, tgtConn)
// Send FIN to client.
clientConn.CloseWrite()
tgtConn.CloseRead()
fromClientErr := <-fromClientErrCh
if fromClientErr != nil {
return onet.NewConnectionError("ERR_RELAY_CLIENT", "Failed to relay traffic from client", fromClientErr)
}
if fromTargetErr != nil {
return onet.NewConnectionError("ERR_RELAY_TARGET", "Failed to relay traffic from target", fromTargetErr)
}
return nil
}
func (h *tcpHandler) handleConnection(ctx context.Context, listenerPort int, clientInfo ipinfo.IPInfo, outerConn transport.StreamConn, proxyMetrics *metrics.ProxyMetrics) (string, *onet.ConnectionError) {
// Set a deadline to receive the address to the target.
readDeadline := time.Now().Add(h.readTimeout)
if deadline, ok := ctx.Deadline(); ok {
outerConn.SetDeadline(deadline)
if deadline.Before(readDeadline) {
readDeadline = deadline
}
}
outerConn.SetReadDeadline(readDeadline)
id, innerConn, authErr := h.authenticate(outerConn)
if authErr != nil {
// Drain to protect against probing attacks.
h.absorbProbe(listenerPort, outerConn, authErr.Status, proxyMetrics)
return id, authErr
}
h.m.AddAuthenticatedTCPConnection(outerConn.RemoteAddr(), id)
// Read target address and dial it.
tgtAddr, err := getProxyRequest(innerConn)
// Clear the deadline for the target address
outerConn.SetReadDeadline(time.Time{})
if err != nil {
// Drain to prevent a close on cipher error.
io.Copy(io.Discard, outerConn)
return id, onet.NewConnectionError("ERR_READ_ADDRESS", "Failed to get target address", err)
}
dialer := transport.FuncStreamDialer(func(ctx context.Context, addr string) (transport.StreamConn, error) {
tgtConn, err := h.dialer.DialStream(ctx, tgtAddr)
if err != nil {
return nil, err
}
tgtConn = metrics.MeasureConn(tgtConn, &proxyMetrics.ProxyTarget, &proxyMetrics.TargetProxy)
return tgtConn, nil
})
return id, proxyConnection(ctx, dialer, tgtAddr, innerConn)
}
// Keep the connection open until we hit the authentication deadline to protect against probing attacks
// `proxyMetrics` is a pointer because its value is being mutated by `clientConn`.
func (h *tcpHandler) absorbProbe(listenerPort int, clientConn io.ReadCloser, status string, proxyMetrics *metrics.ProxyMetrics) {
// This line updates proxyMetrics.ClientProxy before it's used in AddTCPProbe.
_, drainErr := io.Copy(io.Discard, clientConn) // drain socket
drainResult := drainErrToString(drainErr)
logger.Debugf("Drain error: %v, drain result: %v", drainErr, drainResult)
h.m.AddTCPProbe(status, drainResult, listenerPort, proxyMetrics.ClientProxy)
}
func drainErrToString(drainErr error) string {
netErr, ok := drainErr.(net.Error)
switch {
case drainErr == nil:
return "eof"
case ok && netErr.Timeout():
return "timeout"
default:
return "other"
}
}
// NoOpTCPMetrics is a [TCPMetrics] that doesn't do anything. Useful in tests
// or if you don't want to track metrics.
type NoOpTCPMetrics struct{}
var _ TCPMetrics = (*NoOpTCPMetrics)(nil)
func (m *NoOpTCPMetrics) AddClosedTCPConnection(clientInfo ipinfo.IPInfo, clientAddr net.Addr, accessKey string, status string, data metrics.ProxyMetrics, duration time.Duration) {
}
func (m *NoOpTCPMetrics) GetIPInfo(net.IP) (ipinfo.IPInfo, error) {
return ipinfo.IPInfo{}, nil
}
func (m *NoOpTCPMetrics) AddOpenTCPConnection(clientInfo ipinfo.IPInfo) {}
func (m *NoOpTCPMetrics) AddAuthenticatedTCPConnection(clientAddr net.Addr, accessKey string) {
}
func (m *NoOpTCPMetrics) AddTCPProbe(status, drainResult string, port int, clientProxyBytes int64) {
}
func (m *NoOpTCPMetrics) AddTCPCipherSearch(accessKeyFound bool, timeToCipher time.Duration) {}