Repository: knative-sandbox/kn-plugin-func
Pull request #770 (2022-01-20 21:25:53)

Enable custom CA for specified server names
Allow usage of custom Certification Authority for specified host names.

knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/kn-plugin-func/pull/770#" title="Author self-approved">matejvasek</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Fkn-plugin-func).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/knative-sandbox/kn-plugin-func/blob/main/OWNERS)~~ [matejvasek]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/kn-plugin-func/pull/770?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#770](https://codecov.io/gh/knative-sandbox/kn-plugin-func/pull/770?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (f89958f) into [main](https://codecov.io/gh/knative-sandbox/kn-plugin-func/commit/ce938122d8b621e5a1758f2abce152ff92f1e2c9?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (ce93812) will **increase** coverage by `0.35%`.
> The diff coverage is `71.92%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/kn-plugin-func/pull/770/graphs/tree.svg?width=650&height=150&src=pr&token=da1B9bMsQ9&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/kn-plugin-func/pull/770?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@            Coverage Diff             @@
##             main     #770      +/-   ##
==========================================
+ Coverage   41.46%   41.82%   +0.35%     
==========================================
  Files          44       45       +1     
  Lines        4179     4251      +72     
==========================================
+ Hits         1733     1778      +45     
- Misses       2191     2211      +20     
- Partials      255      262       +7     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/kn-plugin-func/pull/770?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [http/transport.go](https://codecov.io/gh/knative-sandbox/kn-plugin-func/pull/770/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-aHR0cC90cmFuc3BvcnQuZ28=) | `62.50% <71.92%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/kn-plugin-func/pull/770?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/kn-plugin-func/pull/770?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [ce93812...f89958f](https://codecov.io/gh/knative-sandbox/kn-plugin-func/pull/770?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

matejvasek: @knative-sandbox/kn-plugin-func-approvers PTAL
matejvasek: This is needed for OpenShift internal registry push.
knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/kn-plugin-func/pull/770#" title="Author self-approved">matejvasek</a>*, *<a href="https://github.com/knative-sandbox/kn-plugin-func/pull/770#pullrequestreview-859514532" title="Approved">zroubalik</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Fkn-plugin-func).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/knative-sandbox/kn-plugin-func/blob/main/OWNERS)~~ [matejvasek,zroubalik]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
matejvasek: @zroubalik can you lgtm?
knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/kn-plugin-func/pull/770#" title="Author self-approved">matejvasek</a>*, *<a href="https://github.com/knative-sandbox/kn-plugin-func/pull/770#pullrequestreview-859636175" title="LGTM">zroubalik</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Fkn-plugin-func).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/knative-sandbox/kn-plugin-func/blob/main/OWNERS)~~ [matejvasek,zroubalik]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->

Patch:
diff --git a/http/transport.go b/http/transport.go
index 69000303..4a1b197b 100644
--- a/http/transport.go
+++ b/http/transport.go
@@ -2,45 +2,86 @@ package http
 
 import (
 	"context"
+	"crypto/tls"
+	"crypto/x509"
 	"errors"
+	"fmt"
 	"io"
 	"net"
 	"net/http"
-	"sync"
 	"time"
 
 	"knative.dev/kn-plugin-func/k8s"
 )
 
+type ContextDialer interface {
+	DialContext(ctx context.Context, network string, addr string) (net.Conn, error)
+	Close() error
+}
+
 type RoundTripCloser interface {
 	http.RoundTripper
 	io.Closer
 }
 
+type options struct {
+	selectCA        func(ctx context.Context, serverName string) (*x509.Certificate, error)
+	inClusterDialer ContextDialer
+}
+
+type Option func(*options)
+
+func WithSelectCA(selectCA func(ctx context.Context, serverName string) (*x509.Certificate, error)) Option {
+	return func(o *options) {
+		o.selectCA = selectCA
+	}
+}
+
+func WithInClusterDialer(inClusterDialer ContextDialer) Option {
+	return func(o *options) {
+		o.inClusterDialer = inClusterDialer
+	}
+}
+
 // NewRoundTripper returns new closable RoundTripper that first tries to dial connection in standard way,
 // if the dial operation fails due to hostname resolution the RoundTripper tries to dial from in cluster pod.
 //
 // This is useful for accessing cluster internal services (pushing a CloudEvent into Knative broker).
-func NewRoundTripper() RoundTripCloser {
-	result := &roundTripCloser{}
+func NewRoundTripper(opts ...Option) RoundTripCloser {
+	o := options{
+		inClusterDialer: k8s.NewLazyInitInClusterDialer(),
+	}
+	for _, option := range opts {
+		option(&o)
+	}
+
+	httpTransport := newHTTPTransport()
+
+	primaryDialer := dialContextFn(httpTransport.DialContext)
+	secondaryDialer := o.inClusterDialer
+
+	combinedDialer := newDialerWithFallback(primaryDialer, secondaryDialer)
+
+	httpTransport.DialContext = combinedDialer.DialContext
+
+	httpTransport.DialTLSContext = newDialTLSContext(combinedDialer, httpTransport.TLSClientConfig, o.selectCA)
+
+	return &roundTripCloser{
+		Transport: httpTransport,
+		dialer:    combinedDialer,
+	}
+}
+
+func newHTTPTransport() *http.Transport {
 	if dt, ok := http.DefaultTransport.(*http.Transport); ok {
-		d := &dialer{
-			defaultDialContext: dt.DialContext,
-		}
-		result.d = d
-		result.Transport = dt.Clone()
-		result.Transport.DialContext = d.DialContext
+		return dt.Clone()
 	} else {
-		d := &dialer{
-			defaultDialContext: (&net.Dialer{
+		return &http.Transport{
+			Proxy: http.ProxyFromEnvironment,
+			DialContext: (&net.Dialer{
 				Timeout:   30 * time.Second,
 				KeepAlive: 30 * time.Second,
 			}).DialContext,
-		}
-		result.d = d
-		result.Transport = &http.Transport{
-			Proxy:                 http.ProxyFromEnvironment,
-			DialContext:           d.DialContext,
 			ForceAttemptHTTP2:     true,
 			MaxIdleConns:          100,
 			IdleConnTimeout:       90 * time.Second,
@@ -48,27 +89,31 @@ func NewRoundTripper() RoundTripCloser {
 			ExpectContinueTimeout: 1 * time.Second,
 		}
 	}
-
-	return result
 }
 
 type roundTripCloser struct {
 	*http.Transport
-	d *dialer
+	dialer ContextDialer
 }
 
 func (r *roundTripCloser) Close() error {
-	return r.d.Close()
+	return r.dialer.Close()
+}
+
+func newDialerWithFallback(primaryDialer ContextDialer, fallbackDialer ContextDialer) *dialerWithFallback {
+	return &dialerWithFallback{
+		primaryDialer:  primaryDialer,
+		fallbackDialer: fallbackDialer,
+	}
 }
 
-type dialer struct {
-	o                  sync.Once
-	defaultDialContext func(ctx context.Context, network, address string) (net.Conn, error)
-	inClusterDialer    k8s.ContextDialer
+type dialerWithFallback struct {
+	primaryDialer  ContextDialer
+	fallbackDialer ContextDialer
 }
 
-func (d *dialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
-	conn, err := d.defaultDialContext(ctx, network, address)
+func (d *dialerWithFallback) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
+	conn, err := d.primaryDialer.DialContext(ctx, network, address)
 	if err == nil {
 		return conn, nil
 	}
@@ -77,26 +122,73 @@ func (d *dialer) DialContext(ctx context.Context, network, address string) (net.
 	if !(errors.As(err, &dnsErr) && dnsErr.IsNotFound) {
 		return nil, err
 	}
-	err = nil
 
-	d.o.Do(func() {
-		d.inClusterDialer, err = k8s.NewInClusterDialer(ctx)
-	})
+	return d.fallbackDialer.DialContext(ctx, network, address)
+}
+
+func (d *dialerWithFallback) Close() error {
+	var err error
+	errs := make([]error, 0, 2)
 
+	err = d.primaryDialer.Close()
 	if err != nil {
-		return nil, err
+		errs = append(errs, err)
 	}
 
-	if d.inClusterDialer == nil {
-		return nil, errors.New("failed to init in cluster dialer")
+	err = d.fallbackDialer.Close()
+	if err != nil {
+		errs = append(errs, err)
 	}
 
-	return d.inClusterDialer.DialContext(ctx, network, address)
+	if len(errs) > 0 {
+		return fmt.Errorf("failed to Close(): %v", errs)
+	}
+
+	return nil
+}
+
+type dialContextFn func(ctx context.Context, network string, addr string) (net.Conn, error)
+
+func (d dialContextFn) DialContext(ctx context.Context, network string, addr string) (net.Conn, error) {
+	return d(ctx, network, addr)
 }
 
-func (d *dialer) Close() error {
-	if d.inClusterDialer != nil {
-		return d.inClusterDialer.Close()
+func (d dialContextFn) Close() error { return nil }
+
+func newDialTLSContext(dialer ContextDialer, config *tls.Config, selectCA func(ctx context.Context, serverName string) (*x509.Certificate, error)) func(ctx context.Context, network, addr string) (net.Conn, error) {
+	if selectCA == nil {
+		return nil
+	}
+	return func(ctx context.Context, network, addr string) (net.Conn, error) {
+
+		conn, err := dialer.DialContext(ctx, network, addr)
+		if err != nil {
+			return nil, err
+		}
+
+		var cfg *tls.Config
+		if config != nil {
+			cfg = config.Clone()
+		} else {
+			cfg = &tls.Config{}
+		}
+
+		serverName, _, err := net.SplitHostPort(addr)
+		if err != nil {
+			return nil, err
+		}
+
+		if cfg.ServerName == "" {
+			cfg.ServerName = serverName
+		}
+
+		if ca, err := selectCA(ctx, serverName); ca != nil && err == nil {
+			caPool := x509.NewCertPool()
+			caPool.AddCert(ca)
+			cfg.RootCAs = caPool
+		}
+
+		tlsConn := tls.Client(conn, cfg)
+		return tlsConn, nil
 	}
-	return nil
 }
diff --git a/k8s/dialer.go b/k8s/dialer.go
index 263d1bf6..163ff0f9 100644
--- a/k8s/dialer.go
+++ b/k8s/dialer.go
@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"sync"
 	"time"
 
 	coreV1 "k8s.io/api/core/v1"
@@ -25,11 +26,6 @@ const (
 	socatImage = "alpine/socat:1.7.4.2-r0"
 )
 
-type ContextDialer interface {
-	DialContext(ctx context.Context, network string, addr string) (net.Conn, error)
-	Close() error
-}
-
 // NewInClusterDialer creates context dialer that will dial TCP connections via POD running in k8s cluster.
 // This is useful when accessing k8s services that are not exposed outside cluster (e.g. openshift image registry).
 //
@@ -48,7 +44,7 @@ type ContextDialer interface {
 //     var client = http.Client{
 //         Transport: transport,
 //     }
-func NewInClusterDialer(ctx context.Context) (ContextDialer, error) {
+func NewInClusterDialer(ctx context.Context) (*contextDialer, error) {
 	c := &contextDialer{
 		detachChan: make(chan struct{}),
 	}
@@ -372,3 +368,30 @@ func newConn(execDone <-chan struct{}) (*io.PipeReader, *io.PipeWriter, conn) {
 	rwc := conn{pr: pr0, pw: pw1, execDone: execDone}
 	return pr1, pw0, rwc
 }
+
+func NewLazyInitInClusterDialer() *lazyInitInClusterDialer {
+	return &lazyInitInClusterDialer{}
+}
+
+type lazyInitInClusterDialer struct {
+	contextDialer *contextDialer
+	initErr       error
+	o             sync.Once
+}
+
+func (l *lazyInitInClusterDialer) DialContext(ctx context.Context, network string, addr string) (net.Conn, error) {
+	l.o.Do(func() {
+		l.contextDialer, l.initErr = NewInClusterDialer(ctx)
+	})
+	if l.initErr != nil {
+		return nil, l.initErr
+	}
+	return l.contextDialer.DialContext(ctx, network, addr)
+}
+
+func (l *lazyInitInClusterDialer) Close() error {
+	if l.contextDialer != nil {
+		return l.contextDialer.Close()
+	}
+	return nil
+}


Test patch:
diff --git a/http/transport_test.go b/http/transport_test.go
new file mode 100644
index 00000000..c214cb37
--- /dev/null
+++ b/http/transport_test.go
@@ -0,0 +1,150 @@
+package http_test
+
+import (
+	"context"
+	"crypto/ed25519"
+	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"fmt"
+	"math/big"
+	"net"
+	"net/http"
+	"testing"
+	"time"
+
+	fnhttp "knative.dev/kn-plugin-func/http"
+)
+
+const inClusterHostName = "image-registry.openshift-image-registry.svc"
+
+func TestCustomCA(t *testing.T) {
+	var err error
+	inClusterAddr, inClusterCA := startServer(t, inClusterHostName)
+	localhostAddr, localhostCA := startServer(t, "localhost")
+
+	mockSelectCA := func(ctx context.Context, serverName string) (*x509.Certificate, error) {
+
+		if serverName == inClusterHostName {
+			return inClusterCA, nil
+		}
+		if serverName == "localhost" {
+			return localhostCA, nil
+		}
+		return nil, nil
+	}
+
+	mockInCusterDialer := mockInClusterDialer{
+		backingAddr: inClusterAddr,
+	}
+
+	tr := fnhttp.NewRoundTripper(
+		fnhttp.WithSelectCA(mockSelectCA),
+		fnhttp.WithInClusterDialer(mockInCusterDialer))
+	defer tr.Close()
+
+	client := http.Client{Transport: tr}
+
+	_, p, err := net.SplitHostPort(localhostAddr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	resp, err := client.Get(fmt.Sprintf("https://localhost:%s", p))
+	if err != nil {
+		t.Fatal(err)
+	}
+	resp.Body.Close()
+
+	resp, err = client.Get(fmt.Sprintf("https://%s:5000/v2/", inClusterHostName))
+	if err != nil {
+		t.Fatal(err)
+	}
+	resp.Body.Close()
+
+}
+
+type mockInClusterDialer struct {
+	backingAddr string
+}
+
+func (m mockInClusterDialer) DialContext(ctx context.Context, network string, addr string) (net.Conn, error) {
+	hostname, _, err := net.SplitHostPort(addr)
+	if err != nil {
+		return nil, err
+	}
+	if hostname == inClusterHostName {
+		return net.Dial(network, m.backingAddr)
+	}
+	return net.Dial(network, addr)
+}
+
+func (m mockInClusterDialer) Close() error {
+	return nil
+}
+
+func startServer(t *testing.T, hostname string) (addr string, ca *x509.Certificate) {
+	randReader := rand.Reader
+
+	caPublicKey, caPrivateKey, err := ed25519.GenerateKey(randReader)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ca = &x509.Certificate{
+		SerialNumber: big.NewInt(1),
+		Subject: pkix.Name{
+			CommonName: hostname,
+		},
+		IPAddresses:           []net.IP{net.IPv4(127, 0, 0, 1), net.IPv6loopback},
+		DNSNames:              []string{"localhost", hostname},
+		NotBefore:             time.Now(),
+		NotAfter:              time.Now().AddDate(10, 0, 0),
+		IsCA:                  true,
+		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth, x509.ExtKeyUsageServerAuth},
+		ExtraExtensions:       []pkix.Extension{},
+		KeyUsage:              x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign,
+		BasicConstraintsValid: true,
+	}
+
+	caBytes, err := x509.CreateCertificate(randReader, ca, ca, caPublicKey, caPrivateKey)
+	if err != nil {
+		t.Fatal()
+	}
+
+	ca, err = x509.ParseCertificate(caBytes)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	cert := tls.Certificate{
+		Certificate: [][]byte{caBytes},
+		PrivateKey:  caPrivateKey,
+		Leaf:        ca,
+	}
+
+	listener, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	addr = listener.Addr().String()
+
+	handler := http.HandlerFunc(func(resp http.ResponseWriter, req *http.Request) {
+	})
+
+	server := http.Server{
+		Handler: handler,
+		TLSConfig: &tls.Config{
+			ServerName:   hostname,
+			Certificates: []tls.Certificate{cert},
+		},
+	}
+	t.Cleanup(func() {
+		server.Close()
+	})
+
+	go func() {
+		_ = server.ServeTLS(listener, "", "")
+	}()
+	return
+}
diff --git a/k8s/dialer_test.go b/k8s/dialer_test.go
index f8cbeef3..b9372ecf 100644
--- a/k8s/dialer_test.go
+++ b/k8s/dialer_test.go
@@ -109,10 +109,7 @@ func TestDialInClusterService(t *testing.T) {
 	// wait for service to start
 	time.Sleep(time.Second * 10)
 
-	dialer, err := k8s.NewInClusterDialer(ctx)
-	if err != nil {
-		t.Fatal(err)
-	}
+	dialer := k8s.NewLazyInitInClusterDialer()
 	t.Cleanup(func() {
 		dialer.Close()
 	})