portainer/api/http/proxy/factory/docker.go

package factory

import (
	"io"
	"net/http"
	"strings"

	portainer "github.com/portainer/portainer/api"
	"github.com/portainer/portainer/api/crypto"
	"github.com/portainer/portainer/api/http/proxy/factory/docker"
	"github.com/portainer/portainer/api/internal/endpointutils"
	"github.com/portainer/portainer/api/url"
	httperror "github.com/portainer/portainer/pkg/libhttp/error"
	"github.com/portainer/portainer/pkg/libhttp/ssrf"

	"github.com/rs/zerolog/log"
)

func (factory *ProxyFactory) newDockerProxy(endpoint *portainer.Endpoint) (http.Handler, error) {
	if strings.HasPrefix(endpoint.URL, "unix://") || strings.HasPrefix(endpoint.URL, "npipe://") {
		return factory.newDockerLocalProxy(endpoint)
	}

	return factory.newDockerHTTPProxy(endpoint)
}

func (factory *ProxyFactory) newDockerLocalProxy(endpoint *portainer.Endpoint) (http.Handler, error) {
	endpointURL, err := url.ParseURL(endpoint.URL)
	if err != nil {
		return nil, err
	}

	return factory.newOSBasedLocalProxy(endpointURL.Path, endpoint)
}

func (factory *ProxyFactory) newDockerHTTPProxy(endpoint *portainer.Endpoint) (http.Handler, error) {
	rawURL := endpoint.URL
	if endpoint.Type == portainer.EdgeAgentOnDockerEnvironment {
		tunnelAddr, err := factory.reverseTunnelService.TunnelAddr(endpoint)
		if err != nil {
			return nil, err
		}
		rawURL = "http://" + tunnelAddr
	}

	endpointURL, err := url.ParseURL(rawURL)
	if err != nil {
		return nil, err
	}

	endpointURL.Scheme = "http"

	transportParameters := &docker.TransportParameters{
		Endpoint:             endpoint,
		DataStore:            factory.dataStore,
		ReverseTunnelService: factory.reverseTunnelService,
		SignatureService:     factory.signatureService,
		DockerClientFactory:  factory.dockerClientFactory,
	}

	var innerTransport *http.Transport
	if endpoint.TLSConfig.TLS || endpoint.TLSConfig.TLSSkipVerify {
		tlsConfig, err := crypto.CreateTLSConfigurationFromDisk(endpoint.TLSConfig)
		if err != nil {
			return nil, err
		}

		endpointURL.Scheme = "https"

		if endpointutils.IsEdgeEndpoint(endpoint) {
			innerTransport = ssrf.NewInternalTransport(tlsConfig)
		} else {
			innerTransport = ssrf.NewTransport(tlsConfig)
		}
	} else if endpointutils.IsEdgeEndpoint(endpoint) {
		innerTransport = ssrf.NewInternalTransport(nil)
	} else {
		innerTransport = ssrf.NewTransport(nil)
	}

	dockerTransport, err := docker.NewTransport(transportParameters, innerTransport, factory.gitService, factory.snapshotService)
	if err != nil {
		return nil, err
	}

	proxy := NewSingleHostReverseProxyWithHostHeader(endpointURL)
	proxy.Transport = dockerTransport
	return proxy, nil
}

type dockerLocalProxy struct {
	transport *docker.Transport
}

// ServeHTTP is the http.Handler interface implementation
// for a local (Unix socket or Windows named pipe) Docker proxy.
func (proxy *dockerLocalProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	// Force URL/domain to http/unixsocket to be able to
	// use http.transport RoundTrip to do the requests via the socket
	r.URL.Scheme = "http"
	r.URL.Host = "unixsocket"

	res, err := proxy.transport.ProxyDockerRequest(r)
	if err != nil {
		code := http.StatusInternalServerError
		if res != nil && res.StatusCode != 0 {
			code = res.StatusCode
		}

		httperror.WriteError(w, code, "Unable to proxy the request via the Docker socket", err)
		return
	}
	defer func() {
		if err := res.Body.Close(); err != nil {
			log.Warn().Err(err).Msg("proxy error: failed to close response body")
		}
	}()

	for k, vv := range res.Header {
		for _, v := range vv {
			w.Header().Add(k, v)
		}
	}

	w.WriteHeader(res.StatusCode)

	streamResponse(w, res.Body)
}

// streamResponse copies body to w, flushing after every chunk instead of using
// io.Copy. io.Copy buffers into the http.ResponseWriter (~2KB) and only flushes
// when the buffer fills or the handler returns, so low-throughput streaming
// docker responses (container logs follow, events, stats, attach) would arrive
// in multi-second batches instead of live; flushing per chunk delivers them as
// they are produced. Behaviour otherwise matches io.Copy: the n>0 chunk is
// written BEFORE the readErr check so a simultaneous Read -> (n>0, io.EOF)
// still emits the final chunk; a write error or a non-EOF read error breaks the
// loop with a Debug log; and a writer that does not implement http.Flusher
// degrades to the old buffered behaviour rather than panicking.
func streamResponse(w http.ResponseWriter, body io.Reader) {
	flusher, _ := w.(http.Flusher)
	buf := make([]byte, 32*1024)
	for {
		n, readErr := body.Read(buf)
		if n > 0 {
			if _, writeErr := w.Write(buf[:n]); writeErr != nil {
				log.Debug().Err(writeErr).Msg("proxy error")
				break
			}
			if flusher != nil {
				flusher.Flush()
			}
		}
		if readErr != nil {
			if readErr != io.EOF {
				log.Debug().Err(readErr).Msg("proxy error")
			}
			break
		}
	}
}