#4711

Committed 03 Jun 2024 08:39PM UTC coverage: 59.016% (+0.1%) from 58.918%

Build # #4711

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Make upload client/server certs configurable (#3252)

* Add client cert config to CDI resource

Signed-off-by: Michael Henriksen <mhenriks@redhat.com>

* make client certs configurable

Signed-off-by: Michael Henriksen <mhenriks@redhat.com>

* Create uploadserver.Config

Signed-off-by: Michael Henriksen <mhenriks@redhat.com>

* uploadserver should read certs from files

Signed-off-by: Michael Henriksen <mhenriks@redhat.com>

* make sure to not close doneChan when error occurs

generally tighten up handling of "done" "uploading" and "processing"

Signed-off-by: Michael Henriksen <mhenriks@redhat.com>

* add deadline support to uploadserver

Signed-off-by: Michael Henriksen <mhenriks@redhat.com>

* Add deadline support to upload controller

Signed-off-by: Michael Henriksen <mhenriks@redhat.com>

* clone controller should use configured client cert duration

Signed-off-by: Michael Henriksen <mhenriks@redhat.com>

* make lint check happy

Signed-off-by: Michael Henriksen <mhenriks@redhat.com>

* Extend existing func test to validate client certs configurable and will be rotated

Signed-off-by: Michael Henriksen <mhenriks@redhat.com>

* Use deadline/rotation for clone pods as well

Forgot about the case where a source PVC may be in use.  Bay be a big delay from when target pod is created and source.

Signed-off-by: Michael Henriksen <mhenriks@redhat.com>

---------

Signed-off-by: Michael Henriksen <mhenriks@redhat.com>

Run Details

283 of 371 new or added lines in 12 files covered. (76.28%)

9 existing lines in 4 files now uncovered.

16269 of 27567 relevant lines covered (59.02%)

0.65 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

69.75

/pkg/uploadserver/uploadserver.go

/*
 * This file is part of the CDI project
 *
 * 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
 *
 *     http://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.
 *
 * Copyright 2018 Red Hat, Inc.
 *
 */

package uploadserver

import (
        "archive/tar"
        "context"
        "crypto/tls"
        "crypto/x509"
        "fmt"
        "io"
        "mime/multipart"
        "net"
        "net/http"
        "os"
        "strings"
        "sync"
        "time"

        "github.com/golang/snappy"
        "github.com/pkg/errors"

        "k8s.io/klog/v2"

        cdiv1 "kubevirt.io/containerized-data-importer-api/pkg/apis/core/v1beta1"
        "kubevirt.io/containerized-data-importer/pkg/common"
        "kubevirt.io/containerized-data-importer/pkg/importer"
        "kubevirt.io/containerized-data-importer/pkg/util"
        cryptowatch "kubevirt.io/containerized-data-importer/pkg/util/tls-crypto-watch"
)

const (
        healthzPort = 8080
        healthzPath = "/healthz"
)

type Config struct {
        BindAddress string
        BindPort    int

        Destination string

        ServerKeyFile, ServerCertFile string
        ClientCertFile, ClientName    string

        ImageSize          string
        FilesystemOverhead float64
        Preallocation      bool

        Deadline *time.Time

        CryptoConfig cryptowatch.CryptoConfig
}

// RunResult is the result of the upload server run
type RunResult struct {
        CloneTarget          bool
        PreallocationApplied bool
        DeadlinePassed       bool
}

// UploadServer is the interface to uploadServerApp
type UploadServer interface {
        Run() (*RunResult, error)
}

type uploadServerApp struct {
        config               *Config
        mux                  *http.ServeMux
        uploading            bool
        processing           bool
        done                 bool
        preallocationApplied bool
        cloneTarget          bool
        doneChan             chan struct{}
        errChan              chan error
        mutex                sync.Mutex
}

type imageReadCloser func(*http.Request) (io.ReadCloser, error)

// may be overridden in tests
var uploadProcessorFunc = newUploadStreamProcessor
var uploadProcessorFuncAsync = newAsyncUploadStreamProcessor

func bodyReadCloser(r *http.Request) (io.ReadCloser, error) {
        return r.Body, nil
}

func formReadCloser(r *http.Request) (io.ReadCloser, error) {
        multiReader, err := r.MultipartReader()
        if err != nil {
                return nil, err
        }

        var filePart *multipart.Part

        for {
                filePart, err = multiReader.NextPart()
                if err != nil || filePart.FormName() == "file" {
                        break
                }
                klog.Infof("Ignoring part %s", filePart.FormName())
        }

        // multiReader.NextPart() returns io.EOF when read everything
        if err != nil {
                return nil, err
        }

        return filePart, nil
}

func isCloneTraget(contentType string) bool {
        return contentType == common.BlockdeviceClone || contentType == common.FilesystemCloneContentType
}

// NewUploadServer returns a new instance of uploadServerApp
func NewUploadServer(config *Config) UploadServer {
        server := &uploadServerApp{
                config:    config,
                mux:       http.NewServeMux(),
                uploading: false,
                done:      false,
                doneChan:  make(chan struct{}),
                errChan:   make(chan error),
        }

        for _, path := range common.SyncUploadPaths {
                server.mux.HandleFunc(path, server.uploadHandler(bodyReadCloser))
        }
        for _, path := range common.AsyncUploadPaths {
                server.mux.HandleFunc(path, server.uploadHandlerAsync(bodyReadCloser))
        }
        for _, path := range common.ArchiveUploadPaths {
                server.mux.HandleFunc(path, server.uploadArchiveHandler(bodyReadCloser))
        }
        for _, path := range common.SyncUploadFormPaths {
                server.mux.HandleFunc(path, server.uploadHandler(formReadCloser))
        }
        for _, path := range common.AsyncUploadFormPaths {
                server.mux.HandleFunc(path, server.uploadHandlerAsync(formReadCloser))
        }

        return server
}

func (app *uploadServerApp) Run() (*RunResult, error) {
        uploadServer := http.Server{
                Handler:           app,
                ReadHeaderTimeout: 10 * time.Second,
        }

        healthzServer, err := app.createHealthzServer()
        if err != nil {
                return nil, errors.Wrap(err, "Error creating healthz http server")
        }

        uploadListener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", app.config.BindAddress, app.config.BindPort))
        if err != nil {
                return nil, errors.Wrap(err, "Error creating upload listerner")
        }

        healthzListener, err := net.Listen("tcp", fmt.Sprintf(":%d", healthzPort))
        if err != nil {
                return nil, errors.Wrap(err, "Error creating healthz listerner")
        }

        tlsConfig, err := app.getTLSConfig()
        if err != nil {
                return nil, errors.Wrap(err, "Error getting TLS config")
        }

        go func() {
                defer uploadListener.Close()

                // maybe bind port was 0 (unit tests) assign port here
                app.config.BindPort = uploadListener.Addr().(*net.TCPAddr).Port

                if tlsConfig != nil {
                        uploadServer.TLSConfig = tlsConfig
                        app.errChan <- uploadServer.ServeTLS(uploadListener, "", "")
                        return
                }

                // not sure we want to support this code path
                app.errChan <- uploadServer.Serve(uploadListener)
        }()

        go func() {
                defer healthzServer.Close()

                app.errChan <- healthzServer.Serve(healthzListener)
        }()

        var timeChan <-chan time.Time

        if app.config.Deadline != nil {
                timeChan = time.After(time.Until(*app.config.Deadline))
        } else {
                tc := make(chan time.Time)
                defer close(tc)
                timeChan = tc
        }

        select {
        case err = <-app.errChan:
                if err != nil {
                        klog.Errorf("HTTP server returned error %s", err.Error())
                        return nil, err
                }
        case <-app.doneChan:
                klog.Info("Shutting down http server after successful upload")
                if err := healthzServer.Shutdown(context.Background()); err != nil {
                        klog.Errorf("failed to shutdown healthzServer; %v", err)
                }
                if err := uploadServer.Shutdown(context.Background()); err != nil {
                        klog.Errorf("failed to shutdown uploadServer; %v", err)
                }
        case <-timeChan:
                klog.Info("deadline exceeded, shutting down")
                app.mutex.Lock()
                defer app.mutex.Unlock()
                for {
                        if app.uploading || app.processing {
                                klog.Info("waiting for upload to finish")
                                app.mutex.Unlock()
                                time.Sleep(2 * time.Second)
                                app.mutex.Lock()
                        } else {
                                break
                        }
                }
                if !app.done {
                        klog.Info("upload not done, process exiting")
                        return &RunResult{DeadlinePassed: true}, nil
                }
        }

        result := &RunResult{
                CloneTarget:          app.cloneTarget,
                PreallocationApplied: app.preallocationApplied,
        }

        return result, nil
}

func (app *uploadServerApp) getTLSConfig() (*tls.Config, error) {
        if app.config.ServerCertFile == "" || app.config.ServerKeyFile == "" {
                return nil, nil
        }

        //nolint:gosec // False positive: Min version is not known statically
        config := &tls.Config{
                CipherSuites: app.config.CryptoConfig.CipherSuites,
                ClientAuth:   tls.RequireAndVerifyClientCert,
                MinVersion:   app.config.CryptoConfig.MinVersion,
        }

        if app.config.ClientCertFile != "" {
                bs, err := os.ReadFile(app.config.ClientCertFile)
                if err != nil {
                        return nil, err
                }

                caCertPool := x509.NewCertPool()
                if ok := caCertPool.AppendCertsFromPEM(bs); !ok {
                        return nil, err
                }

                config.ClientCAs = caCertPool
        }

        cert, err := tls.LoadX509KeyPair(app.config.ServerCertFile, app.config.ServerKeyFile)
        if err != nil {
                return nil, err
        }

        config.Certificates = []tls.Certificate{cert}

        return config, nil
}

func (app *uploadServerApp) createHealthzServer() (*http.Server, error) {
        mux := http.NewServeMux()
        mux.HandleFunc(healthzPath, app.healthzHandler)
        return &http.Server{
                Handler:           mux,
                ReadHeaderTimeout: 10 * time.Second,
        }, nil
}

func (app *uploadServerApp) ServeHTTP(w http.ResponseWriter, r *http.Request) {
        app.mux.ServeHTTP(w, r)
}

func (app *uploadServerApp) healthzHandler(w http.ResponseWriter, r *http.Request) {
        if _, err := io.WriteString(w, "OK"); err != nil {
                klog.Errorf("healthzHandler: failed to send response; %v", err)
        }
}

func (app *uploadServerApp) validateShouldHandleRequest(w http.ResponseWriter, r *http.Request) bool {
        if r.Method != http.MethodPost {
                w.WriteHeader(http.StatusNotFound)
                return false
        }

        if r.TLS != nil {
                found := false

                for _, cert := range r.TLS.PeerCertificates {
                        if cert.Subject.CommonName == app.config.ClientName {
                                found = true
                                break
                        }
                }

                if !found {
                        w.WriteHeader(http.StatusUnauthorized)
                        return false
                }
        } else {
                klog.V(3).Infof("Handling HTTP connection")
        }

        app.mutex.Lock()
        defer app.mutex.Unlock()

        if app.uploading || app.processing {
                klog.Warning("Got concurrent upload request")
                w.WriteHeader(http.StatusServiceUnavailable)
                return false
        }

        if app.done {
                klog.Warning("Got upload request after already done")
                w.WriteHeader(http.StatusConflict)
                return false
        }

        app.uploading = true

        return true
}

func (app *uploadServerApp) uploadHandlerAsync(irc imageReadCloser) http.HandlerFunc {
        return func(w http.ResponseWriter, r *http.Request) {
                if r.Method == http.MethodHead {
                        w.WriteHeader(http.StatusOK)
                        return
                }

                if !app.validateShouldHandleRequest(w, r) {
                        return
                }

                cdiContentType := r.Header.Get(common.UploadContentTypeHeader)

                klog.Infof("Content type header is %q\n", cdiContentType)

                readCloser, err := irc(r)
                if err != nil {
                        w.WriteHeader(http.StatusBadRequest)
                }

                processor, err := uploadProcessorFuncAsync(readCloser, app.config.Destination, app.config.ImageSize, app.config.FilesystemOverhead, app.config.Preallocation, cdiContentType)

                app.mutex.Lock()
                defer app.mutex.Unlock()

                if err != nil {
                        klog.Errorf("Saving stream failed: %s", err)
                        if errors.As(err, &importer.ValidationSizeError{}) {
                                w.WriteHeader(http.StatusBadRequest)
                        } else {
                                w.WriteHeader(http.StatusInternalServerError)
                        }

                        _, writeErr := fmt.Fprintf(w, "Saving stream failed: %s", err.Error())
                        if writeErr != nil {
                                klog.Errorf("failed to send response; %v", err)
                        }

                        app.uploading = false
                        return
                }

                app.uploading = false
                app.processing = true

                // Start processing.
                go func() {
                        err := processor.ProcessDataResume()
                        app.mutex.Lock()
                        defer app.mutex.Unlock()
                        app.processing = false
                        if err != nil {
                                klog.Errorf("Error during resumed processing: %v", err)
                                app.errChan <- err
                                return
                        }
                        defer close(app.doneChan)
                        app.done = true
                        app.preallocationApplied = processor.PreallocationApplied()
                        app.cloneTarget = isCloneTraget(cdiContentType)
                        klog.Infof("Wrote data to %s", app.config.Destination)
                }()

                klog.Info("Returning success to caller, continue processing in background")
        }
}

func (app *uploadServerApp) processUpload(irc imageReadCloser, w http.ResponseWriter, r *http.Request, dvContentType cdiv1.DataVolumeContentType) {
        if !app.validateShouldHandleRequest(w, r) {
                return
        }

        cdiContentType := r.Header.Get(common.UploadContentTypeHeader)

        klog.Infof("Content type header is %q\n", cdiContentType)

        readCloser, err := irc(r)
        if err != nil {
                w.WriteHeader(http.StatusBadRequest)
        }

        preallocationApplied, err := uploadProcessorFunc(readCloser, app.config.Destination, app.config.ImageSize, app.config.FilesystemOverhead, app.config.Preallocation, cdiContentType, dvContentType)

        app.mutex.Lock()
        defer app.mutex.Unlock()
        app.uploading = false

        if err != nil {
                klog.Errorf("Saving stream failed: %s", err)
                w.WriteHeader(http.StatusInternalServerError)
                return
        }

        app.done = true
        app.preallocationApplied = preallocationApplied
        app.cloneTarget = isCloneTraget(cdiContentType)
        close(app.doneChan)

        if dvContentType == cdiv1.DataVolumeArchive {
                klog.Infof("Wrote archive data")
        } else {
                klog.Infof("Wrote data to %s", app.config.Destination)
        }
}

func (app *uploadServerApp) uploadHandler(irc imageReadCloser) http.HandlerFunc {
        return func(w http.ResponseWriter, r *http.Request) {
                app.processUpload(irc, w, r, cdiv1.DataVolumeKubeVirt)
        }
}

func (app *uploadServerApp) uploadArchiveHandler(irc imageReadCloser) http.HandlerFunc {
        return func(w http.ResponseWriter, r *http.Request) {
                app.processUpload(irc, w, r, cdiv1.DataVolumeArchive)
        }
}

func newAsyncUploadStreamProcessor(stream io.ReadCloser, dest, imageSize string, filesystemOverhead float64, preallocation bool, sourceContentType string) (*importer.DataProcessor, error) {
        if sourceContentType == common.FilesystemCloneContentType {
                return nil, fmt.Errorf("async filesystem clone not supported")
        }

        uds := importer.NewAsyncUploadDataSource(newContentReader(stream, sourceContentType))
        processor := importer.NewDataProcessor(uds, dest, common.ImporterVolumePath, common.ScratchDataDir, imageSize, filesystemOverhead, preallocation, "")
        return processor, processor.ProcessDataWithPause()
}

func newUploadStreamProcessor(stream io.ReadCloser, dest, imageSize string, filesystemOverhead float64, preallocation bool, sourceContentType string, dvContentType cdiv1.DataVolumeContentType) (bool, error) {
        if sourceContentType == common.FilesystemCloneContentType {
                return false, filesystemCloneProcessor(stream, dest)
        }

        // Clone block device to block device or file system
        uds := importer.NewUploadDataSource(newContentReader(stream, sourceContentType), dvContentType)
        processor := importer.NewDataProcessor(uds, dest, common.ImporterVolumePath, common.ScratchDataDir, imageSize, filesystemOverhead, preallocation, "")
        err := processor.ProcessData()
        return processor.PreallocationApplied(), err
}

// Clone file system to block device or file system
func filesystemCloneProcessor(stream io.ReadCloser, dest string) error {
        // Clone to block device
        if dest == common.WriteBlockPath {
                if err := untarToBlockdev(newSnappyReadCloser(stream), dest); err != nil {
                        return errors.Wrapf(err, "error unarchiving to %s", dest)
                }
                return nil
        }

        // Clone to file system
        destDir := common.ImporterVolumePath
        if err := util.UnArchiveTar(newSnappyReadCloser(stream), destDir); err != nil {
                return errors.Wrapf(err, "error unarchiving to %s", destDir)
        }
        return nil
}

func untarToBlockdev(stream io.Reader, dest string) error {
        tr := tar.NewReader(stream)
        for {
                header, err := tr.Next()
                switch {
                case err == io.EOF:
                        return nil
                case err != nil:
                        return err
                case header == nil:
                        continue
                }
                if !strings.Contains(header.Name, common.DiskImageName) {
                        continue
                }
                switch header.Typeflag {
                case tar.TypeReg, tar.TypeGNUSparse:
                        klog.Infof("Untaring %d bytes to %s", header.Size, dest)
                        f, err := os.OpenFile(dest, os.O_APPEND|os.O_WRONLY, os.ModeDevice|os.ModePerm)
                        if err != nil {
                                return err
                        }
                        written, err := io.CopyN(f, tr, header.Size)
                        if err != nil {
                                return err
                        }
                        klog.Infof("Written %d", written)
                        f.Close()
                        return nil
                }
        }
}

func newContentReader(stream io.ReadCloser, contentType string) io.ReadCloser {
        if contentType == common.BlockdeviceClone {
                return newSnappyReadCloser(stream)
        }

        return stream
}

func newSnappyReadCloser(stream io.ReadCloser) io.ReadCloser {
        return io.NopCloser(snappy.NewReader(stream))
}

1	/*
2	* This file is part of the CDI project
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*
16	* Copyright 2018 Red Hat, Inc.
17	*
18	*/
19
20	package uploadserver
21
22	import (
23	"archive/tar"
24	"context"
25	"crypto/tls"
26	"crypto/x509"
27	"fmt"
28	"io"
29	"mime/multipart"
30	"net"
31	"net/http"
32	"os"
33	"strings"
34	"sync"
35	"time"
36
37	"github.com/golang/snappy"
38	"github.com/pkg/errors"
39
40	"k8s.io/klog/v2"
41
42	cdiv1 "kubevirt.io/containerized-data-importer-api/pkg/apis/core/v1beta1"
43	"kubevirt.io/containerized-data-importer/pkg/common"
44	"kubevirt.io/containerized-data-importer/pkg/importer"
45	"kubevirt.io/containerized-data-importer/pkg/util"
46	cryptowatch "kubevirt.io/containerized-data-importer/pkg/util/tls-crypto-watch"
47	)
48
49	const (
50	healthzPort = 8080
51	healthzPath = "/healthz"
52	)
53
54	type Config struct {
55	BindAddress string
56	BindPort int
57
58	Destination string
59
60	ServerKeyFile, ServerCertFile string
61	ClientCertFile, ClientName string
62
63	ImageSize string
64	FilesystemOverhead float64
65	Preallocation bool
66
67	Deadline *time.Time
68
69	CryptoConfig cryptowatch.CryptoConfig
70	}
71
72	// RunResult is the result of the upload server run
73	type RunResult struct {
74	CloneTarget bool
75	PreallocationApplied bool
76	DeadlinePassed bool
77	}
78
79	// UploadServer is the interface to uploadServerApp
80	type UploadServer interface {
81	Run() (*RunResult, error)
82	}
83
84	type uploadServerApp struct {
85	config *Config
86	mux *http.ServeMux
87	uploading bool
88	processing bool
89	done bool
90	preallocationApplied bool
91	cloneTarget bool
92	doneChan chan struct{}
93	errChan chan error
94	mutex sync.Mutex
95	}
96
97	type imageReadCloser func(*http.Request) (io.ReadCloser, error)
98
99	// may be overridden in tests
100	var uploadProcessorFunc = newUploadStreamProcessor
101	var uploadProcessorFuncAsync = newAsyncUploadStreamProcessor
102
103	func bodyReadCloser(r *http.Request) (io.ReadCloser, error) {	1✔
104	return r.Body, nil	1✔
105	}	1✔
106
107	func formReadCloser(r *http.Request) (io.ReadCloser, error) {	1✔
108	multiReader, err := r.MultipartReader()	1✔
109	if err != nil {	1✔
110	return nil, err	×
111	}	×
112
113	var filePart *multipart.Part	1✔
114		1✔
115	for {	2✔
116	filePart, err = multiReader.NextPart()	1✔
117	if err != nil \|\| filePart.FormName() == "file" {	2✔
118	break	1✔
119	}
120	klog.Infof("Ignoring part %s", filePart.FormName())	×
121	}
122
123	// multiReader.NextPart() returns io.EOF when read everything
124	if err != nil {	1✔
125	return nil, err	×
126	}	×
127
128	return filePart, nil	1✔
129	}
130
131	func isCloneTraget(contentType string) bool {	1✔
132	return contentType == common.BlockdeviceClone \|\| contentType == common.FilesystemCloneContentType	1✔
133	}	1✔
134
135	// NewUploadServer returns a new instance of uploadServerApp
136	func NewUploadServer(config *Config) UploadServer {	1✔
137	server := &uploadServerApp{	1✔
138	config: config,	1✔
139	mux: http.NewServeMux(),	1✔
140	uploading: false,	1✔
141	done: false,	1✔
142	doneChan: make(chan struct{}),	1✔
143	errChan: make(chan error),	1✔
144	}	1✔
145		1✔
146	for _, path := range common.SyncUploadPaths {	2✔
147	server.mux.HandleFunc(path, server.uploadHandler(bodyReadCloser))	1✔
148	}	1✔
149	for _, path := range common.AsyncUploadPaths {	2✔
150	server.mux.HandleFunc(path, server.uploadHandlerAsync(bodyReadCloser))	1✔
151	}	1✔
152	for _, path := range common.ArchiveUploadPaths {	2✔
153	server.mux.HandleFunc(path, server.uploadArchiveHandler(bodyReadCloser))	1✔
154	}	1✔
155	for _, path := range common.SyncUploadFormPaths {	2✔
156	server.mux.HandleFunc(path, server.uploadHandler(formReadCloser))	1✔
157	}	1✔
158	for _, path := range common.AsyncUploadFormPaths {	2✔
159	server.mux.HandleFunc(path, server.uploadHandlerAsync(formReadCloser))	1✔
160	}	1✔
161
162	return server	1✔
163	}
164
165	func (app uploadServerApp) Run() (RunResult, error) {	1✔
166	uploadServer := http.Server{	1✔
167	Handler: app,	1✔
168	ReadHeaderTimeout: 10 * time.Second,	1✔
169	}	1✔
170		1✔
171	healthzServer, err := app.createHealthzServer()	1✔
172	if err != nil {	1✔
NEW 173	return nil, errors.Wrap(err, "Error creating healthz http server")	×
174	}	×
175
176	uploadListener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", app.config.BindAddress, app.config.BindPort))	1✔
177	if err != nil {	1✔
NEW 178	return nil, errors.Wrap(err, "Error creating upload listerner")	×
179	}	×
180
181	healthzListener, err := net.Listen("tcp", fmt.Sprintf(":%d", healthzPort))	1✔
182	if err != nil {	1✔
NEW 183	return nil, errors.Wrap(err, "Error creating healthz listerner")	×
NEW 184	}	×
185
186	tlsConfig, err := app.getTLSConfig()	1✔
187	if err != nil {	1✔
NEW 188	return nil, errors.Wrap(err, "Error getting TLS config")	×
UNCOV 189	}	×
190
191	go func() {	2✔
192	defer uploadListener.Close()	1✔
193		1✔
194	// maybe bind port was 0 (unit tests) assign port here	1✔
195	app.config.BindPort = uploadListener.Addr().(*net.TCPAddr).Port	1✔
196		1✔
197	if tlsConfig != nil {	2✔
198	uploadServer.TLSConfig = tlsConfig	1✔
199	app.errChan <- uploadServer.ServeTLS(uploadListener, "", "")	1✔
200	return	1✔
201	}	1✔
202
203	// not sure we want to support this code path
204	app.errChan <- uploadServer.Serve(uploadListener)	×
205	}()
206
207	go func() {	2✔
208	defer healthzServer.Close()	1✔
209		1✔
210	app.errChan <- healthzServer.Serve(healthzListener)	1✔
211	}()	1✔
212
213	var timeChan <-chan time.Time	1✔
214		1✔
215	if app.config.Deadline != nil {	2✔
216	timeChan = time.After(time.Until(*app.config.Deadline))	1✔
217	} else {	2✔
218	tc := make(chan time.Time)	1✔
219	defer close(tc)	1✔
220	timeChan = tc	1✔
221	}	1✔
222
223	select {	1✔
224	case err = <-app.errChan:	×
NEW 225	if err != nil {	×
NEW 226	klog.Errorf("HTTP server returned error %s", err.Error())	×
NEW 227	return nil, err	×
NEW 228	}	×
229	case <-app.doneChan:	1✔
230	klog.Info("Shutting down http server after successful upload")	1✔
231	if err := healthzServer.Shutdown(context.Background()); err != nil {	1✔
232	klog.Errorf("failed to shutdown healthzServer; %v", err)	×
233	}	×
234	if err := uploadServer.Shutdown(context.Background()); err != nil {	1✔
235	klog.Errorf("failed to shutdown uploadServer; %v", err)	×
236	}	×
237	case <-timeChan:	1✔
238	klog.Info("deadline exceeded, shutting down")	1✔
239	app.mutex.Lock()	1✔
240	defer app.mutex.Unlock()	1✔
241	for {	2✔
242	if app.uploading \|\| app.processing {	1✔
NEW 243	klog.Info("waiting for upload to finish")	×
NEW 244	app.mutex.Unlock()	×
NEW 245	time.Sleep(2 * time.Second)	×
NEW 246	app.mutex.Lock()	×
247	} else {	1✔
248	break	1✔
249	}
250	}
251	if !app.done {	2✔
252	klog.Info("upload not done, process exiting")	1✔
253	return &RunResult{DeadlinePassed: true}, nil	1✔
254	}	1✔
255	}
256
257	result := &RunResult{	1✔
258	CloneTarget: app.cloneTarget,	1✔
259	PreallocationApplied: app.preallocationApplied,	1✔
260	}	1✔
261		1✔
262	return result, nil	1✔
263	}
264
265	func (app uploadServerApp) getTLSConfig() (tls.Config, error) {	1✔
266	if app.config.ServerCertFile == "" \|\| app.config.ServerKeyFile == "" {	1✔
NEW 267	return nil, nil	×
NEW 268	}	×
269
270	//nolint:gosec // False positive: Min version is not known statically
271	config := &tls.Config{	1✔
272	CipherSuites: app.config.CryptoConfig.CipherSuites,	1✔
273	ClientAuth: tls.RequireAndVerifyClientCert,	1✔
274	MinVersion: app.config.CryptoConfig.MinVersion,	1✔
275	}	1✔
276		1✔
277	if app.config.ClientCertFile != "" {	2✔
278	bs, err := os.ReadFile(app.config.ClientCertFile)	1✔
279	if err != nil {	1✔
NEW 280	return nil, err	×
281	}	×
282
283	caCertPool := x509.NewCertPool()	1✔
284	if ok := caCertPool.AppendCertsFromPEM(bs); !ok {	1✔
NEW 285	return nil, err	×
UNCOV 286	}	×
287
288	config.ClientCAs = caCertPool	1✔
289	}
290
291	cert, err := tls.LoadX509KeyPair(app.config.ServerCertFile, app.config.ServerKeyFile)	1✔
292	if err != nil {	1✔
NEW 293	return nil, err	×
UNCOV 294	}	×
295
296	config.Certificates = []tls.Certificate{cert}	1✔
297		1✔
298	return config, nil	1✔
299	}
300
301	func (app uploadServerApp) createHealthzServer() (http.Server, error) {	1✔
302	mux := http.NewServeMux()	1✔
303	mux.HandleFunc(healthzPath, app.healthzHandler)	1✔
304	return &http.Server{	1✔
305	Handler: mux,	1✔
306	ReadHeaderTimeout: 10 * time.Second,	1✔
307	}, nil	1✔
308	}	1✔
309
310	func (app uploadServerApp) ServeHTTP(w http.ResponseWriter, r http.Request) {	1✔
311	app.mux.ServeHTTP(w, r)	1✔
312	}	1✔
313
314	func (app uploadServerApp) healthzHandler(w http.ResponseWriter, r http.Request) {	1✔
315	if _, err := io.WriteString(w, "OK"); err != nil {	1✔
316	klog.Errorf("healthzHandler: failed to send response; %v", err)	×
317	}	×
318	}
319
320	func (app uploadServerApp) validateShouldHandleRequest(w http.ResponseWriter, r http.Request) bool {	1✔
321	if r.Method != http.MethodPost {	2✔
322	w.WriteHeader(http.StatusNotFound)	1✔
323	return false	1✔
324	}	1✔
325
326	if r.TLS != nil {	2✔
327	found := false	1✔
328		1✔
329	for _, cert := range r.TLS.PeerCertificates {	2✔
330	if cert.Subject.CommonName == app.config.ClientName {	2✔
331	found = true	1✔
332	break	1✔
333	}
334	}
335
336	if !found {	2✔
337	w.WriteHeader(http.StatusUnauthorized)	1✔
338	return false	1✔
339	}	1✔
340	} else {	1✔
341	klog.V(3).Infof("Handling HTTP connection")	1✔
342	}	1✔
343
344	app.mutex.Lock()	1✔
345	defer app.mutex.Unlock()	1✔
346		1✔
347	if app.uploading \|\| app.processing {	2✔
348	klog.Warning("Got concurrent upload request")	1✔
349	w.WriteHeader(http.StatusServiceUnavailable)	1✔
350	return false	1✔
351	}	1✔
352
353	if app.done {	2✔
354	klog.Warning("Got upload request after already done")	1✔
355	w.WriteHeader(http.StatusConflict)	1✔
356	return false	1✔
357	}	1✔
358
359	app.uploading = true	1✔
360		1✔
361	return true	1✔
362	}
363
364	func (app *uploadServerApp) uploadHandlerAsync(irc imageReadCloser) http.HandlerFunc {	1✔
365	return func(w http.ResponseWriter, r *http.Request) {	2✔
366	if r.Method == http.MethodHead {	2✔
367	w.WriteHeader(http.StatusOK)	1✔
368	return	1✔
369	}	1✔
370
371	if !app.validateShouldHandleRequest(w, r) {	2✔
372	return	1✔
373	}	1✔
374
375	cdiContentType := r.Header.Get(common.UploadContentTypeHeader)	1✔
376		1✔
377	klog.Infof("Content type header is %q\n", cdiContentType)	1✔
378		1✔
379	readCloser, err := irc(r)	1✔
380	if err != nil {	1✔
381	w.WriteHeader(http.StatusBadRequest)	×
382	}	×
383
384	processor, err := uploadProcessorFuncAsync(readCloser, app.config.Destination, app.config.ImageSize, app.config.FilesystemOverhead, app.config.Preallocation, cdiContentType)	1✔
385		1✔
386	app.mutex.Lock()	1✔
387	defer app.mutex.Unlock()	1✔
388		1✔
389	if err != nil {	2✔
390	klog.Errorf("Saving stream failed: %s", err)	1✔
391	if errors.As(err, &importer.ValidationSizeError{}) {	1✔
392	w.WriteHeader(http.StatusBadRequest)	×
393	} else {	1✔
394	w.WriteHeader(http.StatusInternalServerError)	1✔
395	}	1✔
396
397	_, writeErr := fmt.Fprintf(w, "Saving stream failed: %s", err.Error())	1✔
398	if writeErr != nil {	1✔
399	klog.Errorf("failed to send response; %v", err)	×
400	}	×
401
402	app.uploading = false	1✔
403	return	1✔
404	}
405
406	app.uploading = false	1✔
407	app.processing = true	1✔
408		1✔
409	// Start processing.	1✔
410	go func() {	2✔
411	err := processor.ProcessDataResume()	1✔
412	app.mutex.Lock()	1✔
413	defer app.mutex.Unlock()	1✔
414	app.processing = false	1✔
415	if err != nil {	1✔
NEW 416	klog.Errorf("Error during resumed processing: %v", err)	×
NEW 417	app.errChan <- err	×
NEW 418	return	×
NEW 419	}	×
420	defer close(app.doneChan)	1✔
421	app.done = true	1✔
422	app.preallocationApplied = processor.PreallocationApplied()	1✔
423	app.cloneTarget = isCloneTraget(cdiContentType)	1✔
424	klog.Infof("Wrote data to %s", app.config.Destination)	1✔
425	}()
426
427	klog.Info("Returning success to caller, continue processing in background")	1✔
428	}
429	}
430
431	func (app uploadServerApp) processUpload(irc imageReadCloser, w http.ResponseWriter, r http.Request, dvContentType cdiv1.DataVolumeContentType) {	1✔
432	if !app.validateShouldHandleRequest(w, r) {	2✔
433	return	1✔
434	}	1✔
435
436	cdiContentType := r.Header.Get(common.UploadContentTypeHeader)	1✔
437		1✔
438	klog.Infof("Content type header is %q\n", cdiContentType)	1✔
439		1✔
440	readCloser, err := irc(r)	1✔
441	if err != nil {	1✔
442	w.WriteHeader(http.StatusBadRequest)	×
443	}	×
444
445	preallocationApplied, err := uploadProcessorFunc(readCloser, app.config.Destination, app.config.ImageSize, app.config.FilesystemOverhead, app.config.Preallocation, cdiContentType, dvContentType)	1✔
446		1✔
447	app.mutex.Lock()	1✔
448	defer app.mutex.Unlock()	1✔
449	app.uploading = false	1✔
450		1✔
451	if err != nil {	2✔
452	klog.Errorf("Saving stream failed: %s", err)	1✔
453	w.WriteHeader(http.StatusInternalServerError)	1✔
454	return	1✔
455	}	1✔
456
457	app.done = true	1✔
458	app.preallocationApplied = preallocationApplied	1✔
459	app.cloneTarget = isCloneTraget(cdiContentType)	1✔
460	close(app.doneChan)	1✔
461		1✔
462	if dvContentType == cdiv1.DataVolumeArchive {	1✔
463	klog.Infof("Wrote archive data")	×
464	} else {	1✔
465	klog.Infof("Wrote data to %s", app.config.Destination)	1✔
466	}	1✔
467	}
468
469	func (app *uploadServerApp) uploadHandler(irc imageReadCloser) http.HandlerFunc {	1✔
470	return func(w http.ResponseWriter, r *http.Request) {	2✔
471	app.processUpload(irc, w, r, cdiv1.DataVolumeKubeVirt)	1✔
472	}	1✔
473	}
474
475	func (app *uploadServerApp) uploadArchiveHandler(irc imageReadCloser) http.HandlerFunc {	1✔
476	return func(w http.ResponseWriter, r *http.Request) {	2✔
477	app.processUpload(irc, w, r, cdiv1.DataVolumeArchive)	1✔
478	}	1✔
479	}
480
481	func newAsyncUploadStreamProcessor(stream io.ReadCloser, dest, imageSize string, filesystemOverhead float64, preallocation bool, sourceContentType string) (*importer.DataProcessor, error) {	×
482	if sourceContentType == common.FilesystemCloneContentType {	×
483	return nil, fmt.Errorf("async filesystem clone not supported")	×
484	}	×
485
486	uds := importer.NewAsyncUploadDataSource(newContentReader(stream, sourceContentType))	×
487	processor := importer.NewDataProcessor(uds, dest, common.ImporterVolumePath, common.ScratchDataDir, imageSize, filesystemOverhead, preallocation, "")	×
488	return processor, processor.ProcessDataWithPause()	×
489	}
490
491	func newUploadStreamProcessor(stream io.ReadCloser, dest, imageSize string, filesystemOverhead float64, preallocation bool, sourceContentType string, dvContentType cdiv1.DataVolumeContentType) (bool, error) {	×
492	if sourceContentType == common.FilesystemCloneContentType {	×
493	return false, filesystemCloneProcessor(stream, dest)	×
494	}	×
495
496	// Clone block device to block device or file system
497	uds := importer.NewUploadDataSource(newContentReader(stream, sourceContentType), dvContentType)	×
498	processor := importer.NewDataProcessor(uds, dest, common.ImporterVolumePath, common.ScratchDataDir, imageSize, filesystemOverhead, preallocation, "")	×
499	err := processor.ProcessData()	×
500	return processor.PreallocationApplied(), err	×
501	}
502
503	// Clone file system to block device or file system
504	func filesystemCloneProcessor(stream io.ReadCloser, dest string) error {	×
505	// Clone to block device	×
506	if dest == common.WriteBlockPath {	×
507	if err := untarToBlockdev(newSnappyReadCloser(stream), dest); err != nil {	×
508	return errors.Wrapf(err, "error unarchiving to %s", dest)	×
509	}	×
510	return nil	×
511	}
512
513	// Clone to file system
514	destDir := common.ImporterVolumePath	×
515	if err := util.UnArchiveTar(newSnappyReadCloser(stream), destDir); err != nil {	×
516	return errors.Wrapf(err, "error unarchiving to %s", destDir)	×
517	}	×
518	return nil	×
519	}
520
521	func untarToBlockdev(stream io.Reader, dest string) error {	×
522	tr := tar.NewReader(stream)	×
523	for {	×
524	header, err := tr.Next()	×
525	switch {	×
526	case err == io.EOF:	×
527	return nil	×
528	case err != nil:	×
529	return err	×
530	case header == nil:	×
531	continue	×
532	}
533	if !strings.Contains(header.Name, common.DiskImageName) {	×
534	continue	×
535	}
536	switch header.Typeflag {	×
537	case tar.TypeReg, tar.TypeGNUSparse:	×
538	klog.Infof("Untaring %d bytes to %s", header.Size, dest)	×
539	f, err := os.OpenFile(dest, os.O_APPEND\|os.O_WRONLY, os.ModeDevice\|os.ModePerm)	×
540	if err != nil {	×
541	return err	×
542	}	×
543	written, err := io.CopyN(f, tr, header.Size)	×
544	if err != nil {	×
545	return err	×
546	}	×
547	klog.Infof("Written %d", written)	×
548	f.Close()	×
549	return nil	×
550	}
551	}
552	}
553
554	func newContentReader(stream io.ReadCloser, contentType string) io.ReadCloser {	×
555	if contentType == common.BlockdeviceClone {	×
556	return newSnappyReadCloser(stream)	×
557	}	×
558
559	return stream	×
560	}
561
562	func newSnappyReadCloser(stream io.ReadCloser) io.ReadCloser {	×
563	return io.NopCloser(snappy.NewReader(stream))	×
564	}	×

kubevirt / containerized-data-importer / #4711

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous