1
package connector
2
 
3
import (
4
	"fmt"
5
	"io"
6
	"net"
7
 
8
	"go.miragespace.co/specter/spec/protocol"
9
	"go.miragespace.co/specter/spec/rpc"
10
	"go.miragespace.co/specter/spec/tun"
11
	"go.miragespace.co/specter/tun/client/dialer"
12
 
13
	"go.uber.org/zap"
14
)
15
 
16
func statusExchange(rw io.ReadWriter) (*protocol.TunnelStatus, error) {
17
	// because of quic's early connection, the client need to "poke" the gateway before
18
	// the gateway can actually accept a stream, despite .OpenStreamSync
19
	status := &protocol.TunnelStatus{}
20
	err := rpc.Send(rw, status)
21
	if err != nil {
22
		return nil, fmt.Errorf("error sending status check: %w", err)
23
	}
24
	status.Reset()
25
	err = rpc.BoundedReceive(rw, status, 1024)
26
	if err != nil {
27
		return nil, fmt.Errorf("error receiving checking status: %w", err)
28
	}
29
	return status, nil
30
}
31
 
32
func GetConnection(d dialer.TransportDialer) (net.Conn, error) {
33
	rw, err := d.Dial()
34
	if err != nil {
35
		return nil, err
36
	}
37
	status, err := statusExchange(rw)
38
	if err != nil {
39
		return nil, err
40
	}
41
	if status.GetStatus() != protocol.TunnelStatusCode_STATUS_OK {
42
		return nil, fmt.Errorf("error opening tunnel: %s", status.Error)
43
	}
44
	return rw, nil
45
}
46
 
47
func HandleConnections(logger *zap.Logger, listener net.Listener, d dialer.TransportDialer) {
48
	for {
49
		conn, err := listener.Accept()
50
		if err != nil {
51
			return
52
		}
53
		go func(local *net.TCPConn) {
54
			r, err := GetConnection(d)
55
			if err != nil {
56
				logger.Error("Error forwarding local connections via specter gateway", zap.Error(err))
57
				local.Close()
58
				return
59
			}
60
 
61
			logger.Info("Forwarding incoming connection", zap.String("local", conn.RemoteAddr().String()), zap.String("via", d.Remote().String()))
62
 
63
			go func() {
64
				errChan := tun.Pipe(r, local)
65
				for err := range errChan {
66
					logger.Error("error piping to target", zap.Error(err))
67
				}
68
			}()
69
		}(conn.(*net.TCPConn))
70
	}
71
}