Forward auth headers to volume servers

Author: mitchellgoffpc

src/lib.go

@@ -143,11 +143,18 @@ func needs_rebalance(volumes []string, kvolumes []string) bool {
 
 // *** Remote Access Functions ***
 
-func remote_delete(remote string) error {
+func addAuth(req *http.Request, auth string) {
+	if auth != "" {
+		req.Header.Set("Authorization", auth)
+	}
+}
+
+func remote_delete(remote string, auth string) error {
 	req, err := http.NewRequest("DELETE", remote, nil)
 	if err != nil {
 		return err
 	}
+	addAuth(req, auth)
 	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
 		return err
@@ -159,12 +166,13 @@ func remote_delete(remote string) error {
 	return nil
 }
 
-func remote_put(remote string, length int64, body io.Reader) error {
+func remote_put(remote string, length int64, body io.Reader, auth string) error {
 	req, err := http.NewRequest("PUT", remote, body)
 	if err != nil {
 		return err
 	}
 	req.ContentLength = length
+	addAuth(req, auth)
 	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
 		return err
@@ -176,8 +184,13 @@ func remote_put(remote string, length int64, body io.Reader) error {
 	return nil
 }
 
-func remote_get(remote string) (string, error) {
-	resp, err := http.Get(remote)
+func remote_get(remote string, auth string) (string, error) {
+	req, err := http.NewRequest("GET", remote, nil)
+	if err != nil {
+		return "", err
+	}
+	addAuth(req, auth)
+	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
 		return "", err
 	}
@@ -192,13 +205,14 @@ func remote_get(remote string) (string, error) {
 	return string(body), nil
 }
 
-func remote_head(remote string, timeout time.Duration) (bool, error) {
+func remote_head(remote string, timeout time.Duration, auth string) (bool, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
 	req, err := http.NewRequestWithContext(ctx, "HEAD", remote, nil)
 	if err != nil {
 		return false, err
 	}
+	addAuth(req, auth)
 	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
 		return false, err

src/rebalance.go

@@ -11,6 +11,7 @@ type RebalanceRequest struct {
 	key      []byte
 	volumes  []string
 	kvolumes []string
+	auth     string
 }
 
 func rebalance(a *App, req RebalanceRequest) bool {
@@ -20,7 +21,7 @@ func rebalance(a *App, req RebalanceRequest) bool {
 	rvolumes := make([]string, 0)
 	for _, rv := range req.volumes {
 		remote_test := fmt.Sprintf("http://%s%s", rv, kp)
-		found, err := remote_head(remote_test, 1*time.Minute)
+		found, err := remote_head(remote_test, 1*time.Minute, req.auth)
 		if err != nil {
 			fmt.Println("rebalance head error", err, remote_test)
 			return false
@@ -49,7 +50,7 @@ func rebalance(a *App, req RebalanceRequest) bool {
 		remote_from := fmt.Sprintf("http://%s%s", v, kp)
 
 		// read
-		ss, err = remote_get(remote_from)
+		ss, err = remote_get(remote_from, req.auth)
 		if err != nil {
 			fmt.Println("rebalance get error", err, remote_from)
 		} else {
@@ -74,7 +75,7 @@ func rebalance(a *App, req RebalanceRequest) bool {
 		if needs_write {
 			remote_to := fmt.Sprintf("http://%s%s", v, kp)
 			// write
-			if err := remote_put(remote_to, int64(len(ss)), strings.NewReader(ss)); err != nil {
+			if err := remote_put(remote_to, int64(len(ss)), strings.NewReader(ss), req.auth); err != nil {
 				fmt.Println("rebalance put error", err, remote_to)
 				rebalance_error = true
 			}
@@ -102,7 +103,7 @@ func rebalance(a *App, req RebalanceRequest) bool {
 		}
 		if needs_delete {
 			remote_del := fmt.Sprintf("http://%s%s", v2, kp)
-			if err := remote_delete(remote_del); err != nil {
+			if err := remote_delete(remote_del, req.auth); err != nil {
 				fmt.Println("rebalance delete error", err, remote_del)
 				delete_error = true
 			}

src/rebuild.go

@@ -25,7 +25,7 @@ type RebuildRequest struct {
 func get_files(url string) []File {
 	//fmt.Println(url)
 	var files []File
-	dat, err := remote_get(url)
+	dat, err := remote_get(url, "")
 	if err != nil {
 		fmt.Println("ugh", err)
 		return files

src/server.go

@@ -102,7 +102,7 @@ func (a *App) QueryHandler(key []byte, w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-func (a *App) Delete(key []byte, unlink bool) int {
+func (a *App) Delete(key []byte, unlink bool, auth string) int {
 	// delete the key, first locally
 	rec := a.GetRecord(key)
 	if rec.deleted == HARD || (unlink && rec.deleted == SOFT) {
@@ -123,7 +123,7 @@ func (a *App) Delete(key []byte, unlink bool) int {
 		delete_error := false
 		for _, volume := range rec.rvolumes {
 			remote := fmt.Sprintf("http://%s%s", volume, key2path(key))
-			if remote_delete(remote) != nil {
+			if remote_delete(remote, auth) != nil {
 				// if this fails, it's possible to get an orphan file
 				// but i'm not really sure what else to do?
 				delete_error = true
@@ -142,7 +142,7 @@ func (a *App) Delete(key []byte, unlink bool) int {
 	return 204
 }
 
-func (a *App) WriteToReplicas(key []byte, value io.Reader, valuelen int64) int {
+func (a *App) WriteToReplicas(key []byte, value io.Reader, valuelen int64, auth string) int {
 	// we don't have the key, compute the remote URL
 	kvolumes := key2volume(key, a.volumes, a.replicas, a.subvolumes, a.vdir_colocation)
 
@@ -160,7 +160,7 @@ func (a *App) WriteToReplicas(key []byte, value io.Reader, valuelen int64) int {
 			body = bytes.NewReader(buf.Bytes())
 		}
 		remote := fmt.Sprintf("http://%s%s", kvolumes[i], key2path(key))
-		if remote_put(remote, valuelen, body) != nil {
+		if remote_put(remote, valuelen, body, auth) != nil {
 			// we assume the remote wrote nothing if it failed
 			fmt.Printf("replica %d write failed: %s\n", i, remote)
 			// try not to leave key in INIT (writing) state (ignore errors)
@@ -188,6 +188,7 @@ func (a *App) WriteToReplicas(key []byte, value io.Reader, valuelen int64) int {
 func (a *App) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	key := []byte(r.URL.Path)
 	lkey := []byte(r.URL.Path + r.URL.Query().Get("partNumber"))
+	auth := r.Header.Get("Authorization")
 
 	log.Println(r.Method, r.URL, r.ContentLength, r.Header["Range"])
 
@@ -237,7 +238,7 @@ func (a *App) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			good := false
 			for _, vn := range rand.Perm(len(rec.rvolumes)) {
 				remote = fmt.Sprintf("http://%s%s", rec.rvolumes[vn], key2path(key))
-				found, _ := remote_head(remote, a.voltimeout)
+				found, _ := remote_head(remote, a.voltimeout, auth)
 				if found {
 					good = true
 					break
@@ -285,7 +286,7 @@ func (a *App) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 			for _, subkey := range del.Keys {
 				fullkey := fmt.Sprintf("%s/%s", key, subkey)
-				status := a.Delete([]byte(fullkey), false)
+				status := a.Delete([]byte(fullkey), false, auth)
 				if status != 204 {
 					w.WriteHeader(status)
 					return
@@ -324,7 +325,7 @@ func (a *App) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 				fs = append(fs, f)
 			}
 
-			status := a.WriteToReplicas(key, io.MultiReader(fs...), sz)
+			status := a.WriteToReplicas(key, io.MultiReader(fs...), sz, auth)
 			w.WriteHeader(status)
 			w.Write([]byte("<CompleteMultipartUploadResult></CompleteMultipartUploadResult>"))
 			return
@@ -363,11 +364,11 @@ func (a *App) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			io.Copy(f, r.Body)
 			w.WriteHeader(200)
 		} else {
-			status := a.WriteToReplicas(key, r.Body, r.ContentLength)
+			status := a.WriteToReplicas(key, r.Body, r.ContentLength, auth)
 			w.WriteHeader(status)
 		}
 	case "DELETE", "UNLINK":
-		status := a.Delete(key, r.Method == "UNLINK")
+		status := a.Delete(key, r.Method == "UNLINK", auth)
 		w.WriteHeader(status)
 	case "RELINK":
 		rec := a.GetRecord(key)
@@ -391,7 +392,7 @@ func (a *App) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		}
 
 		kvolumes := key2volume(key, a.volumes, a.replicas, a.subvolumes, a.vdir_colocation)
-		rbreq := RebalanceRequest{key: key, volumes: rec.rvolumes, kvolumes: kvolumes}
+		rbreq := RebalanceRequest{key: key, volumes: rec.rvolumes, kvolumes: kvolumes, auth: auth}
 		if !rebalance(a, rbreq) {
 			w.WriteHeader(400)
 			return