ipfsproxy: intercept block/put and dag/put and pin to cluster on pin=true

This fixes #1738. Tests still missing

api/ipfsproxy/headers.go

@@ -119,7 +119,7 @@ func (proxy *Server) copyHeadersFromIPFSWithRequest(
 	hdrs []string,
 	dest http.Header, req *http.Request,
 ) error {
-	res, err := proxy.ipfsRoundTripper.RoundTrip(req)
+	res, err := proxy.reverseProxy.Transport.RoundTrip(req)
 	if err != nil {
 		logger.Error("error making request for header extraction to ipfs: ", err)
 		return err

api/ipfsproxy/ipfsproxy.go

@@ -31,8 +31,8 @@ import (
 	cmd "github.com/ipfs/go-ipfs-cmds"
 	logging "github.com/ipfs/go-log/v2"
 	path "github.com/ipfs/go-path"
-	peer "github.com/libp2p/go-libp2p/core/peer"
 	rpc "github.com/libp2p/go-libp2p-gorpc"
+	peer "github.com/libp2p/go-libp2p/core/peer"
 	madns "github.com/multiformats/go-multiaddr-dns"
 	manet "github.com/multiformats/go-multiaddr/net"
 
@@ -67,7 +67,7 @@ type Server struct {
 
 	listeners    []net.Listener         // proxy listener
 	server       *http.Server           // proxy server
-	ipfsRoundTripper http.RoundTripper // allows to talk to IPFS
+	reverseProxy *httputil.ReverseProxy // allows to talk to IPFS
 
 	ipfsHeadersStore sync.Map
 
@@ -206,7 +206,7 @@ func New(cfg *Config) (*Server, error) {
 		rpcReady:     make(chan struct{}, 1),
 		listeners:    listeners,
 		server:       s,
-		ipfsRoundTripper: reverseProxy.Transport,
+		reverseProxy: reverseProxy,
 	}
 
 	// Ideally, we should only intercept POST requests, but
@@ -260,6 +260,14 @@ func New(cfg *Config) (*Server, error) {
 		Path("/repo/gc").
 		HandlerFunc(proxy.repoGCHandler).
 		Name("RepoGC")
+	hijackSubrouter.
+		Path("/block/put").
+		HandlerFunc(proxy.blockPutHandler).
+		Name("BlockPut")
+	hijackSubrouter.
+		Path("/dag/put").
+		HandlerFunc(proxy.dagPutHandler).
+		Name("DagPut")
 
 	// Everything else goes to the IPFS daemon.
 	router.PathPrefix("/").Handler(reverseProxy)
@@ -760,6 +768,162 @@ func (proxy *Server) repoGCHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+type ipfsBlockPutResp struct {
+	Key  api.Cid
+	Size int
+}
+
+func (proxy *Server) blockPutHandler(w http.ResponseWriter, r *http.Request) {
+	if r.URL.Query().Get("pin") != "true" {
+		proxy.reverseProxy.ServeHTTP(w, r)
+		return
+	}
+
+	u2, err := url.Parse(proxy.nodeAddr)
+	if err != nil {
+		logger.Error(err)
+		ipfsErrorResponder(w, err.Error(), -1)
+		return
+	}
+
+	r.URL.Host = u2.Host
+	r.URL.Scheme = u2.Scheme
+	r.Host = u2.Host
+	r.RequestURI = ""
+
+	res, err := proxy.reverseProxy.Transport.RoundTrip(r)
+	if err != nil {
+		ipfsErrorResponder(w, err.Error(), -1)
+		return
+	}
+
+	defer res.Body.Close()
+
+	if res.StatusCode != http.StatusOK {
+		w.WriteHeader(res.StatusCode)
+		_, err = io.Copy(w, res.Body)
+		if err != nil {
+			logger.Error(err)
+		}
+		return
+	}
+
+	// Returned 200. Parse responses.
+	w.Header().Set("Trailer", "X-Stream-Error")
+	w.WriteHeader(http.StatusOK) // any errors from here go into trailers
+
+	dec := json.NewDecoder(res.Body)
+	enc := json.NewEncoder(w)
+	for {
+		var res ipfsBlockPutResp
+		err = dec.Decode(&res)
+		if err == io.EOF {
+			return
+		}
+		if err != nil {
+			logger.Error(err)
+			w.Header().Add("X-Stream-Error", err.Error())
+			return
+		}
+		p := api.PinCid(res.Key)
+		var pinObj api.Pin
+		if err := proxy.rpcClient.Call(
+			"",
+			"Cluster",
+			"Pin",
+			p,
+			&pinObj,
+		); err != nil {
+			logger.Error(err)
+			w.Header().Add("X-Stream-Error", err.Error())
+			// keep going though blocks
+		}
+		if err := enc.Encode(res); err != nil {
+			logger.Error(err)
+			w.Header().Add("X-Stream-Error", err.Error())
+			return
+		}
+	}
+}
+
+type ipfsDagPutResp struct {
+	Cid cid.Cid
+}
+
+func (proxy *Server) dagPutHandler(w http.ResponseWriter, r *http.Request) {
+	// Note this mostly duplicates blockPutHandler
+	if r.URL.Query().Get("pin") != "true" {
+		proxy.reverseProxy.ServeHTTP(w, r)
+		return
+	}
+
+	u2, err := url.Parse(proxy.nodeAddr)
+	if err != nil {
+		logger.Error(err)
+		ipfsErrorResponder(w, err.Error(), -1)
+		return
+	}
+
+	r.URL.Host = u2.Host
+	r.URL.Scheme = u2.Scheme
+	r.Host = u2.Host
+	r.RequestURI = ""
+
+	res, err := proxy.reverseProxy.Transport.RoundTrip(r)
+	if err != nil {
+		ipfsErrorResponder(w, err.Error(), -1)
+		return
+	}
+
+	defer res.Body.Close()
+
+	if res.StatusCode != http.StatusOK {
+		w.WriteHeader(res.StatusCode)
+		_, err = io.Copy(w, res.Body)
+		if err != nil {
+			logger.Error(err)
+		}
+		return
+	}
+
+	// Returned 200. Parse responses.
+	w.Header().Set("Trailer", "X-Stream-Error")
+	w.WriteHeader(http.StatusOK) // any errors from here go into trailers
+
+	dec := json.NewDecoder(res.Body)
+	enc := json.NewEncoder(w)
+	for {
+		var res ipfsDagPutResp
+		err = dec.Decode(&res)
+		if err == io.EOF {
+			return
+		}
+		if err != nil {
+			logger.Error(err)
+			w.Header().Add("X-Stream-Error", err.Error())
+			return
+		}
+		p := api.PinCid(api.NewCid(res.Cid))
+		var pinObj api.Pin
+		if err := proxy.rpcClient.Call(
+			"",
+			"Cluster",
+			"Pin",
+			p,
+			&pinObj,
+		); err != nil {
+			logger.Error(err)
+			w.Header().Add("X-Stream-Error", err.Error())
+			// keep going though blocks
+		}
+		if err := enc.Encode(res); err != nil {
+			logger.Error(err)
+			w.Header().Add("X-Stream-Error", err.Error())
+			return
+		}
+	}
+}
+
 // slashHandler returns a handler which converts a /a/b/c/<argument> request
 // into an /a/b/c/<argument>?arg=<argument> one. And uses the given origHandler
 // for it. Our handlers expect that arguments are passed in the ?arg query