-
Notifications
You must be signed in to change notification settings - Fork 0
Add WithDialer option and rewrite With* to auto-inject dialer #29
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
cd9a52a
32261d5
891684b
5c2bb7f
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -22,13 +22,18 @@ import ( | |
|
|
||
| var log = slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{AddSource: true})) | ||
|
|
||
| // DialContextFunc is the canonical dialer type used throughout kindling and its transports. | ||
| type DialContextFunc func(ctx context.Context, network, addr string) (net.Conn, error) | ||
|
|
||
| // Kindling is the interface that wraps the basic Dial and DialContext methods for control | ||
| // plane traffic. | ||
| type Kindling interface { | ||
| // NewHTTPClient returns a new HTTP client that is configured to use kindling. | ||
| NewHTTPClient() *http.Client | ||
| // ReplaceTransport replaces an existing transport RoundTripper generator with the provided one. | ||
| ReplaceTransport(name string, rt func(ctx context.Context, addr string) (http.RoundTripper, error)) error | ||
| // Close releases resources held by transports created by kindling. | ||
| Close() error | ||
| } | ||
| type roundTripperGenerator func(ctx context.Context, addr string) (http.RoundTripper, error) | ||
|
|
||
|
|
@@ -40,6 +45,33 @@ type kindling struct { | |
| logWriter io.Writer | ||
| panicListener func(string) | ||
| appName string // The name of the tool using kindling, used for logging and debugging. | ||
| dialContext DialContextFunc | ||
| closersMu sync.Mutex | ||
| closers []io.Closer | ||
| } | ||
|
|
||
| // closerFunc adapts a func() into an io.Closer. | ||
| type closerFunc func() error | ||
|
|
||
| func (f closerFunc) Close() error { return f() } | ||
|
|
||
| // streamConnAdapter wraps a net.Conn as a transport.StreamConn by adding CloseWrite/CloseRead methods. | ||
| type streamConnAdapter struct { | ||
| net.Conn | ||
| } | ||
|
|
||
| func (s *streamConnAdapter) CloseWrite() error { | ||
| if cw, ok := s.Conn.(interface{ CloseWrite() error }); ok { | ||
| return cw.CloseWrite() | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (s *streamConnAdapter) CloseRead() error { | ||
| if cr, ok := s.Conn.(interface{ CloseRead() error }); ok { | ||
| return cr.CloseRead() | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // Make sure that kindling implements the Kindling interface. | ||
|
|
@@ -50,6 +82,7 @@ var _ Kindling = &kindling{} | |
| const ( | ||
| priorityLogWriter = iota | ||
| priorityPanicListener | ||
| priorityDialer | ||
| ) | ||
|
|
||
| // Option is a functional option type that allows us to configure the Client. | ||
|
|
@@ -73,9 +106,31 @@ func NewKindling(name string, options ...Option) Kindling { | |
| for _, opt := range options { | ||
| opt.apply(k) | ||
| } | ||
|
|
||
| // Default the dialer if none was provided. | ||
| if k.dialContext == nil { | ||
| k.dialContext = (&net.Dialer{}).DialContext | ||
| } | ||
| return k | ||
| } | ||
|
|
||
| // Close releases resources held by transports created by kindling. | ||
| func (k *kindling) Close() error { | ||
| k.closersMu.Lock() | ||
| defer k.closersMu.Unlock() | ||
| var errs []error | ||
| for _, c := range k.closers { | ||
| if err := c.Close(); err != nil { | ||
| errs = append(errs, err) | ||
| } | ||
| } | ||
| k.closers = nil | ||
| if len(errs) > 0 { | ||
| return fmt.Errorf("errors closing kindling resources: %v", errs) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // NewHTTPClient implements the Kindling interface. | ||
| func (k *kindling) NewHTTPClient() *http.Client { | ||
| // Create a specialized HTTP transport that concurrently races between fronted and smart dialer. | ||
|
|
@@ -100,50 +155,98 @@ func (k *kindling) ReplaceTransport(name string, rt func(ctx context.Context, ad | |
| return fmt.Errorf("Could not find matching transport: %v", name) | ||
| } | ||
|
|
||
| // WithDomainFronting is a functional option that sets up domain fronting for kindling using | ||
| // the provided fronted.Fronted instance from https://github.com/getlantern/fronted. | ||
| func WithDomainFronting(f fronted.Fronted) Option { | ||
| log.Info("Setting domain fronting") | ||
| if f == nil { | ||
| log.Error("Fronted instance is nil") | ||
| return &emptyOption{} | ||
| } | ||
| return WithTransport(newTransport("fronted", 0, true, func(ctx context.Context, addr string) (http.RoundTripper, error) { | ||
| return f.NewConnectedRoundTripper(ctx, addr) | ||
| })) | ||
| } | ||
|
|
||
| // WithDNSTunnel is a functional option that sets up a DNS tunnel for kindling using the provided | ||
| // [dnstt.DNSTT] instance | ||
| func WithDNSTunnel(d dnstt.DNSTT) Option { | ||
| log.Info("Setting DNS tunnel") | ||
| if d == nil { | ||
| log.Error("DNSTT instance is nil") | ||
| return &emptyOption{} | ||
| } | ||
| return WithTransport(newTransport("dnstt", 0, true, func(ctx context.Context, addr string) (http.RoundTripper, error) { | ||
| return d.NewRoundTripper(ctx, addr) | ||
| })) | ||
| // WithDomainFronting is a functional option that sets up domain fronting for kindling. | ||
| // It accepts fronted.Option parameters and constructs the fronted instance internally, | ||
| // automatically injecting kindling's dialer. | ||
| // | ||
| // Note: this is a breaking change from the previous API which accepted a fronted.Fronted instance. | ||
| // Callers must now pass fronted.Option values instead (e.g., fronted.WithConfigURL(...)). | ||
| func WithDomainFronting(opts ...fronted.Option) Option { | ||
| return newOption(func(k *kindling) { | ||
| log.Info("Setting domain fronting") | ||
| // Prepend our dialer so the caller's options can override if needed. | ||
| allOpts := make([]fronted.Option, 0, len(opts)+1) | ||
| allOpts = append(allOpts, fronted.WithDialer(fronted.DialFunc(k.dialContext))) | ||
| allOpts = append(allOpts, opts...) | ||
| f := fronted.NewFronted(allOpts...) | ||
| k.closers = append(k.closers, closerFunc(func() error { f.Close(); return nil })) | ||
| k.transports = append(k.transports, newTransport("fronted", 0, true, func(ctx context.Context, addr string) (http.RoundTripper, error) { | ||
| return f.NewConnectedRoundTripper(ctx, addr) | ||
| })) | ||
| }) | ||
| } | ||
|
Comment on lines
+164
to
177
|
||
|
|
||
| // WithAMPCache uses the AMP cache for making requests. It adds an 'amp' round tripper from the provided amp.Client. | ||
| func WithAMPCache(c amp.Client) Option { | ||
| log.Info("Setting AMP cache") | ||
| if c == nil { | ||
| log.Error("AMP client is nil") | ||
| return &emptyOption{} | ||
| } | ||
| return WithTransport(newTransport("amp", 6000, false, func(ctx context.Context, addr string) (http.RoundTripper, error) { | ||
| return c.RoundTripper() | ||
| })) | ||
| // WithDNSTunnel is a functional option that sets up a DNS tunnel for kindling. | ||
| // It accepts dnstt.Option parameters and constructs the dnstt instance internally, | ||
| // automatically injecting kindling's dialer. | ||
| // | ||
| // Note: this is a breaking change from the previous API which accepted a dnstt.DNSTT instance. | ||
| // Callers must now pass dnstt.Option values instead (e.g., dnstt.WithDoH(...), dnstt.WithTunnelDomain(...)). | ||
| // | ||
| // If transport creation fails, the error is logged and the transport is silently omitted. | ||
| // Kindling will still function with any remaining transports. | ||
| func WithDNSTunnel(opts ...dnstt.Option) Option { | ||
| return newOption(func(k *kindling) { | ||
| log.Info("Setting DNS tunnel") | ||
| // Prepend our dialer so the caller's options can override if needed. | ||
| allOpts := make([]dnstt.Option, 0, len(opts)+1) | ||
| allOpts = append(allOpts, dnstt.WithDialer(k.dialContext)) | ||
| allOpts = append(allOpts, opts...) | ||
| d, err := dnstt.NewDNSTT(allOpts...) | ||
| if err != nil { | ||
| log.Error("Failed to create DNSTT instance", "error", err) | ||
| return | ||
| } | ||
|
Comment on lines
+195
to
+199
|
||
| k.closers = append(k.closers, d) | ||
| k.transports = append(k.transports, newTransport("dnstt", 0, true, func(ctx context.Context, addr string) (http.RoundTripper, error) { | ||
| return d.NewRoundTripper(ctx, addr) | ||
| })) | ||
| }) | ||
| } | ||
|
Comment on lines
+188
to
+205
|
||
|
|
||
| // WithAMPCache uses the AMP cache for making requests. It accepts an amp.Config and | ||
| // optional amp.Option parameters, constructs the amp client internally, and automatically | ||
| // injects kindling's dialer. | ||
| // | ||
| // Note: this is a breaking change from the previous API which accepted an amp.Client instance. | ||
| // Callers must now pass an amp.Config and optional amp.Option values instead. | ||
| // | ||
| // If client creation fails, the error is logged and the transport is silently omitted. | ||
| // Kindling will still function with any remaining transports. | ||
| func WithAMPCache(cfg amp.Config, opts ...amp.Option) Option { | ||
| return newOption(func(k *kindling) { | ||
| log.Info("Setting AMP cache") | ||
| // Adapt DialContextFunc to amp's dialFunc (func(network, addr string) (net.Conn, error)). | ||
| // Note: amp.WithDialer does not accept a context, so we use context.Background() here. | ||
| // This means caller-level timeouts/cancellation won't propagate to the dial phase; | ||
| // however, the AMP client's own context (below) handles lifecycle cancellation. | ||
| ampDialer := func(network, addr string) (net.Conn, error) { | ||
| return k.dialContext(context.Background(), network, addr) | ||
| } | ||
|
Comment on lines
+223
to
+225
|
||
| allOpts := make([]amp.Option, 0, len(opts)+1) | ||
| allOpts = append(allOpts, amp.WithDialer(ampDialer)) | ||
| allOpts = append(allOpts, opts...) | ||
|
|
||
| ctx, cancel := context.WithCancel(context.Background()) | ||
| c, err := amp.NewClientWithConfig(ctx, cfg, allOpts...) | ||
| if err != nil { | ||
| cancel() | ||
| log.Error("Failed to create AMP client", "error", err) | ||
| return | ||
| } | ||
|
Comment on lines
+230
to
+236
|
||
| k.closers = append(k.closers, closerFunc(func() error { cancel(); return nil })) | ||
| k.transports = append(k.transports, newTransport("amp", 6000, false, func(ctx context.Context, addr string) (http.RoundTripper, error) { | ||
| return c.RoundTripper() | ||
| })) | ||
| }) | ||
|
Comment on lines
+216
to
+241
|
||
| } | ||
|
|
||
| // WithProxyless is a functional option that enables proxyless mode for the Kindling such that | ||
| // it accesses the control plane directly using a variety of proxyless techniques. | ||
| func WithProxyless(domains ...string) Option { | ||
| return newOption(func(k *kindling) { | ||
| slog.Info("Setting proxyless mode") | ||
| smartDialer, err := newSmartHTTPDialerFunc(k.logWriter, domains...) | ||
| smartDialer, err := newSmartHTTPDialerFunc(k.dialContext, k.logWriter, domains...) | ||
| if err != nil { | ||
| log.Error("Failed to create smart dialer", "error", err) | ||
| return | ||
|
|
@@ -197,6 +300,16 @@ func WithTransport(transport Transport) Option { | |
| }) | ||
| } | ||
|
|
||
| // WithDialer sets a custom dialer that will be automatically injected into all transports | ||
| // (fronted, dnstt, amp, smart) created by kindling. This runs at priorityDialer priority | ||
| // so that it is set before transport options are applied. | ||
| func WithDialer(dial DialContextFunc) Option { | ||
| return newOptionWithPriority(func(k *kindling) { | ||
| log.Info("Setting custom dialer") | ||
| k.dialContext = dial | ||
| }, priorityDialer) | ||
| } | ||
|
|
||
| // WithPanicListener is a functional option that sets a panic listener that should be notified | ||
| // whenever any goroutine panics. We set this with a higher priority so that it is set before | ||
| // any other options that may depend on it. | ||
|
|
@@ -212,8 +325,8 @@ func (k *kindling) newRaceTransport() http.RoundTripper { | |
| return newRaceTransport(k.appName, k.panicListener, k.transports...) | ||
| } | ||
|
|
||
| func newSmartHTTPDialerFunc(logWriter io.Writer, domains ...string) (roundTripperGenerator, error) { | ||
| d, err := newSmartDialer(logWriter, domains...) | ||
| func newSmartHTTPDialerFunc(dialContext DialContextFunc, logWriter io.Writer, domains ...string) (roundTripperGenerator, error) { | ||
| d, err := newSmartDialer(dialContext, logWriter, domains...) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to create smart dialer: %v", err) | ||
| } | ||
|
|
@@ -229,9 +342,12 @@ func newSmartHTTPDialerFunc(logWriter io.Writer, domains ...string) (roundTrippe | |
| } | ||
|
|
||
| // NewSmartHTTPTransport creates a new HTTP transport that uses the Outline smart dialer to dial to the | ||
| // specified domains. | ||
| func NewSmartHTTPTransport(logWriter io.Writer, domains ...string) (*http.Transport, error) { | ||
| d, err := newSmartDialer(logWriter, domains...) | ||
| // specified domains. If dialContext is nil, a default net.Dialer is used. | ||
| func NewSmartHTTPTransport(dialContext DialContextFunc, logWriter io.Writer, domains ...string) (*http.Transport, error) { | ||
| if dialContext == nil { | ||
| dialContext = (&net.Dialer{}).DialContext | ||
| } | ||
| d, err := newSmartDialer(dialContext, logWriter, domains...) | ||
| if err != nil { | ||
| log.Error("Failed to create smart dialer", "error", err) | ||
| return nil, fmt.Errorf("failed to create smart dialer: %v", err) | ||
|
|
@@ -259,11 +375,18 @@ func newTransportWithDialContext(dialContext func(ctx context.Context, network, | |
| //go:embed smart_dialer_config.yml | ||
| var embedFS embed.FS | ||
|
|
||
| func newSmartDialer(logWriter io.Writer, domains ...string) (transport.StreamDialer, error) { | ||
| func newSmartDialer(dialContext DialContextFunc, logWriter io.Writer, domains ...string) (transport.StreamDialer, error) { | ||
| streamDialer := transport.FuncStreamDialer(func(ctx context.Context, addr string) (transport.StreamConn, error) { | ||
| conn, err := dialContext(ctx, "tcp", addr) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return &streamConnAdapter{Conn: conn}, nil | ||
| }) | ||
| finder := &smart.StrategyFinder{ | ||
| TestTimeout: 5 * time.Second, | ||
| LogWriter: logWriter, | ||
| StreamDialer: &transport.TCPDialer{}, | ||
| StreamDialer: streamDialer, | ||
| PacketDialer: &transport.UDPDialer{}, | ||
| } | ||
|
|
||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -4,7 +4,9 @@ import ( | |
| "context" | ||
| "fmt" | ||
| "io" | ||
| "net" | ||
| "net/http" | ||
| "sync/atomic" | ||
| "testing" | ||
|
|
||
| "github.com/getlantern/fronted" | ||
|
|
@@ -16,13 +18,13 @@ func TestNewKindling(t *testing.T) { | |
| t.Run("Success", func(t *testing.T) { | ||
| t.Parallel() | ||
|
|
||
| f := fronted.NewFronted( | ||
| fronted.WithConfigURL("https://media.githubusercontent.com/media/getlantern/fronted/refs/heads/main/fronted.yaml.gz")) | ||
| kindling := NewKindling("kindling", | ||
| WithDomainFronting(f), | ||
| k := NewKindling("kindling", | ||
| WithDomainFronting( | ||
| fronted.WithConfigURL("https://media.githubusercontent.com/media/getlantern/fronted/refs/heads/main/fronted.yaml.gz")), | ||
| WithPanicListener(func(string) {}), | ||
| ) | ||
| if kindling == nil { | ||
| defer k.Close() | ||
| if k == nil { | ||
| t.Errorf("NewKindling() = nil; want non-nil Kindling") | ||
| } | ||
| }) | ||
|
|
@@ -224,3 +226,39 @@ func TestKindling_ReplaceTransport(t *testing.T) { | |
| } | ||
| }) | ||
| } | ||
|
|
||
| func TestWithDialer(t *testing.T) { | ||
| t.Parallel() | ||
| var called atomic.Bool | ||
| customDialer := func(ctx context.Context, network, addr string) (net.Conn, error) { | ||
| called.Store(true) | ||
| return (&net.Dialer{}).DialContext(ctx, network, addr) | ||
| } | ||
| k := NewKindling("test-app", | ||
| WithDialer(customDialer), | ||
| ) | ||
| defer k.Close() | ||
| if k == nil { | ||
| t.Fatal("NewKindling() = nil; want non-nil Kindling") | ||
| } | ||
| // Verify the dialer was stored by checking the internal struct. | ||
| ki := k.(*kindling) | ||
| if ki.dialContext == nil { | ||
| t.Fatal("dialContext should not be nil after WithDialer") | ||
| } | ||
| // Verify the custom dialer is actually the one stored by calling it and checking the flag. | ||
| _, _ = ki.dialContext(context.Background(), "tcp", "localhost:0") | ||
| if !called.Load() { | ||
| t.Fatal("custom dialer should have been called") | ||
| } | ||
| } | ||
|
|
||
| func TestWithDialerDefault(t *testing.T) { | ||
| t.Parallel() | ||
| k := NewKindling("test-app") | ||
| defer k.Close() | ||
| ki := k.(*kindling) | ||
| if ki.dialContext == nil { | ||
| t.Fatal("default dialContext should be set when WithDialer is not used") | ||
| } | ||
| } | ||
|
Comment on lines
+230
to
+264
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The closers slice is not protected by a mutex, which could lead to race conditions if Close() is called concurrently with options being applied or if multiple goroutines call Close() simultaneously. The transports slice has a mutex (roundTripperGeneratorsMutex) protecting it, but closers does not. Consider adding synchronization to protect the closers slice and the Close() method, or document that Close() must not be called concurrently.