1
package client
2
 
3
import (
4
	"context"
5
	"crypto/ed25519"
6
	"crypto/rand"
7
	"crypto/tls"
8
	"crypto/x509"
9
	"fmt"
10
	"io"
11
	"net"
12
	"net/http"
13
	"strings"
14
	"time"
15
 
16
	pkiImpl "go.miragespace.co/specter/pki"
17
	"go.miragespace.co/specter/spec/pki"
18
	"go.miragespace.co/specter/spec/protocol"
19
	"go.miragespace.co/specter/spec/transport"
20
	"go.miragespace.co/specter/tun/client/dialer"
21
	"go.miragespace.co/specter/util"
22
 
23
	"go.uber.org/zap"
24
)
25
 
26
type LightweightConfig struct {
27
	Logger    *zap.Logger
28
	Transport transport.ClientTransport
29
	PKIClient protocol.PKIService
30
	Apex      *dialer.ParsedApex
31
	Target    string
32
	Token     string
33
	Output    io.Writer
34
}
35
 
36
type LightweightClient struct {
37
	LightweightConfig
38
	fwd   *forwarder
39
	route route
40
	// The first Open response preserves the URL and home across disconnections.
41
	initialSession *protocol.OpenSessionResponse
42
}
43
 
44
// attachmentError marks a local failure that another connection cannot fix.
45
type attachmentError struct {
46
	error
47
}
48
 
49
func (e *attachmentError) Unwrap() error { return e.error }
50
 
51
func NewLightweightClient(cfg LightweightConfig) (*LightweightClient, error) {
52
	target, err := parseTarget(cfg.Target)
53
	if err != nil {
54
		return nil, err
55
	}
56
	if cfg.Apex == nil || cfg.Transport == nil || cfg.PKIClient == nil {
57
		return nil, fmt.Errorf("apex, transport, and PKI client are required")
58
	}
59
	if cfg.Apex.Host == "" || cfg.Apex.Port < 1 || cfg.Apex.Port > 65535 {
60
		return nil, fmt.Errorf("invalid apex address")
61
	}
62
	if cfg.Logger == nil {
63
		cfg.Logger = zap.NewNop()
64
	}
65
	if cfg.Output == nil {
66
		cfg.Output = io.Discard
67
	}
68
	return &LightweightClient{
69
		LightweightConfig: cfg,
70
		fwd:               newForwarder(cfg.Logger),
71
		route:             route{parsed: target},
72
	}, nil
73
}
74
 
75
func (l *LightweightClient) Run(ctx context.Context) error {
76
	parent := ctx
77
	_, key, err := ed25519.GenerateKey(rand.Reader)
78
	if err != nil {
79
		return err
80
	}
81
	req, err := pkiImpl.CreateRequest(key)
82
	if err != nil {
83
		return err
84
	}
85
	issueCtx, cancel := context.WithTimeout(ctx, 15*time.Second)
86
	certResp, err := l.PKIClient.RequestCertificate(issueCtx, req)
87
	cancel()
88
	if ctx.Err() != nil {
89
		return nil
90
	}
91
	if err != nil {
92
		return fmt.Errorf("obtaining attachment certificate: %w", err)
93
	}
94
	cert, err := x509.ParseCertificate(certResp.GetCertDer())
95
	if err != nil {
96
		return err
97
	}
98
	identity, err := pki.ExtractCertificateIdentity(cert)
99
	if err != nil {
100
		return err
101
	}
102
	notAfter := cert.NotAfter
103
	l.Logger.Info("Attachment certificate obtained", zap.Object("identity", identity), zap.Time("notAfter", notAfter))
104
	if err := l.Transport.WithClientCertificate(tls.Certificate{
105
		Certificate: [][]byte{cert.Raw},
106
		PrivateKey:  key,
107
	}); err != nil {
108
		return err
109
	}
110
	ctx, cancel = context.WithDeadline(ctx, notAfter)
111
	defer cancel()
112
	defer l.fwd.closeAll()
113
	router := transport.NewStreamRouter(l.Logger, nil, l.Transport)
114
	router.HandleTunnel(protocol.Stream_DIRECT, func(d *transport.StreamDelegate) {
115
		link, ok := receiveLink(l.Logger, d)
116
		if ok {
117
			l.fwd.handleLink(ctx, link, d, l.route)
118
		}
119
	})
120
	go router.Accept(ctx)
121
	if l.Token == "" {
122
		err = l.runEphemeral(ctx)
123
	} else {
124
		err = l.runToken(ctx)
125
	}
126
	if parent.Err() != nil {
127
		return nil
128
	}
129
	if ctx.Err() != nil {
130
		return fmt.Errorf("attachment certificate expired at %s", notAfter.Format(time.RFC3339))
131
	}
132
	return err
133
}
134
 
135
func (l *LightweightClient) runEphemeral(ctx context.Context) error {
136
	backoff := time.Second
137
	for ctx.Err() == nil {
138
		node := l.initialSession.GetNode()
139
		if node == nil {
140
			node = &protocol.Node{Address: l.Apex.String()}
141
		}
142
		activated, err := l.attach(ctx, node)
143
		if ctx.Err() != nil {
144
			break
145
		}
146
		if definite(err) {
147
			return err
148
		}
149
		if unsupported(err) {
150
			return fmt.Errorf("server %s does not support lightweight tunnels", node.GetAddress())
151
		}
152
		if activated {
153
			backoff = time.Second
154
		}
155
		l.Logger.Warn("Retrying tunnel attachment", zap.Error(err), zap.Duration("backoff", backoff))
156
		timer := time.NewTimer(util.RandomTimeRange(backoff))
157
		select {
158
		case <-ctx.Done():
159
			timer.Stop()
160
		case <-timer.C:
161
		}
162
		backoff = min(backoff*2, 30*time.Second)
163
	}
164
	return ctx.Err()
165
}
166
 
167
func (l *LightweightClient) dialAttachment(ctx context.Context, node *protocol.Node) (net.Conn, transport.PhysicalConn, error) {
168
	// Overlay retains this context for the physical connection. Only RPC requests
169
	// get attempt deadlines; ending a successful unary request must not end traffic.
170
	conn, err := l.Transport.DialStream(ctx, node, protocol.Stream_RPC)
171
	if err != nil {
172
		return nil, nil, err
173
	}
174
	provider, ok := conn.(transport.PhysicalConnProvider)
175
	if !ok || provider.PhysicalConn() == nil {
176
		conn.Close()
177
		return nil, nil, &attachmentError{fmt.Errorf("transport does not expose a physical connection")}
178
	}
179
	return conn, provider.PhysicalConn(), nil
180
}
181
 
182
// Each client makes one request on the supplied stream; its caller owns the
183
// physical connection. Closing the RPC stream must not end other tunnel streams.
184
func tunnelClientOnStream(conn net.Conn) protocol.TunnelService {
185
	tp := http.DefaultTransport.(*http.Transport).Clone()
186
	tp.DisableKeepAlives = true
187
	tp.DialTLSContext = func(context.Context, string, string) (net.Conn, error) {
188
		return conn, nil
189
	}
190
	return protocol.NewTunnelServiceProtobufClient("https://tunnel", &http.Client{Transport: tp})
191
}
192
 
193
func (l *LightweightClient) attach(ctx context.Context, node *protocol.Node) (bool, error) {
194
	conn, pc, err := l.dialAttachment(ctx, node)
195
	if err != nil {
196
		return false, err
197
	}
198
	defer pc.Close("attachment ended")
199
	defer conn.Close()
200
	openCtx, cancel := context.WithTimeout(ctx, 15*time.Second)
201
	resp, err := tunnelClientOnStream(conn).OpenEphemeralSession(openCtx, &protocol.OpenEphemeralSessionRequest{})
202
	cancel()
203
	if err != nil {
204
		return false, err
205
	}
206
	first := l.initialSession == nil
207
	if err := l.acceptSession(resp); err != nil {
208
		return false, &attachmentError{err}
209
	}
210
	if !first {
211
		l.Logger.Info("Tunnel recovered", zap.String("url", l.URL()), zap.String("server", resp.GetNode().GetAddress()))
212
	}
213
	select {
214
	case <-ctx.Done():
215
	case <-pc.Done():
216
	}
217
	l.Logger.Info("Tunnel disconnected", zap.Error(pc.Err()))
218
	return true, pc.Err()
219
}
220
 
221
func (l *LightweightClient) acceptSession(resp *protocol.OpenSessionResponse) error {
222
	if resp.GetHostname() == "" || resp.GetNode().GetAddress() == "" || resp.GetApex() == "" {
223
		return fmt.Errorf("server returned an empty hostname, identity, or apex")
224
	}
225
	if l.initialSession != nil {
226
		if l.initialSession.GetHostname() != resp.GetHostname() {
227
			return fmt.Errorf("recovered tunnel hostname changed")
228
		}
229
		return nil
230
	}
231
	l.initialSession = resp
232
	l.fwd.rootDomain.Store(resp.GetApex())
233
	if _, err := fmt.Fprintln(l.Output, l.URL()); err != nil {
234
		return &attachmentError{fmt.Errorf("printing tunnel URL: %w", err)}
235
	}
236
	l.Logger.Info("Tunnel ready", zap.String("url", l.URL()), zap.String("hostname", resp.GetHostname()), zap.String("server", resp.GetNode().GetAddress()))
237
	return nil
238
}
239
 
240
func (l *LightweightClient) URL() string {
241
	hostname := l.initialSession.GetHostname()
242
	if strings.Contains(hostname, ".") {
243
		return "https://" + hostname
244
	}
245
	authority := hostname + "." + l.initialSession.GetApex()
246
	if l.Apex.Port != 443 {
247
		authority = net.JoinHostPort(authority, fmt.Sprint(l.Apex.Port))
248
	}
249
	return "https://" + authority
250
}