net (03/12): Add authorized asynchronous node diagnostic APIs

cmd/unbounded-net-controller/detail_aggregated_test.go

@@ -0,0 +1,103 @@
+// Copyright (c) Microsoft Corporation.
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1"
+	"github.com/Azure/unbounded/internal/net/webhook"
+)
+
+func TestAggregatedDetailsUsesTrustedProxyAndSharedLifecycle(t *testing.T) {
+	certPEM, _, caPEM, err := webhook.GenerateClientAuthCertificateForTest("front-proxy-client")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	cert, err := x509.ParseCertificate(mustParseCertPEM(t, certPEM))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	server := testWebhookServerForPush(t, caPEM)
+	manager := testDetailRequests(t, nodeDetailRequestHooks{})
+	health := &healthState{detailRequests: manager, registerAggregatedAPIServer: true}
+	health.isLeader.Store(true)
+
+	mux := http.NewServeMux()
+	registerStatusHandlers(mux, health, true, server, nil, nil)
+
+	path := "/apis/status.net.unbounded-cloud.io/v1alpha1/nodes/node/details"
+
+	send := func(method, path, body string, trusted bool) *httptest.ResponseRecorder {
+		t.Helper()
+
+		request := httptest.NewRequest(method, path, strings.NewReader(body))
+		request.Header.Set("X-Remote-User", "viewer")
+
+		if trusted {
+			request.TLS = &tls.ConnectionState{PeerCertificates: []*x509.Certificate{cert}}
+		}
+
+		recorder := httptest.NewRecorder()
+		mux.ServeHTTP(recorder, request)
+
+		return recorder
+	}
+
+	for _, method := range []string{http.MethodPost, http.MethodGet} {
+		if response := send(method, path, "{}", false); response.Code != http.StatusForbidden {
+			t.Fatalf("spoofed front-proxy header accepted: %d", response.Code)
+		}
+	}
+
+	response := send(http.MethodPost, path, `{"forceRefresh":true}`, true)
+	if response.Code != http.StatusAccepted {
+		t.Fatalf("aggregated request failed: %d %s", response.Code, response.Body.String())
+	}
+
+	var pending statusv1alpha1.NodeDetailResult
+	if err := json.Unmarshal(response.Body.Bytes(), &pending); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := manager.Complete("node", pending.RequestID, testDetailStatus()); err != nil {
+		t.Fatal(err)
+	}
+
+	for _, resultPath := range []string{path, "/status/node/node/details"} {
+		response = send(http.MethodGet, resultPath+"?requestId="+pending.RequestID, "", true)
+
+		var result statusv1alpha1.NodeDetailResult
+		if err := json.Unmarshal(response.Body.Bytes(), &result); err != nil {
+			t.Fatal(err)
+		}
+
+		if response.Code != http.StatusOK || result.State != statusv1alpha1.NodeDetailComplete ||
+			result.RequestID != pending.RequestID || result.Details == nil || result.Details.Status.NodeInfo.Name != "node" {
+			t.Fatalf("paths do not share detail lifecycle: %d %+v", response.Code, result)
+		}
+	}
+
+	if response = send(http.MethodDelete, path, "", true); response.Code != http.StatusMethodNotAllowed {
+		t.Fatalf("unsupported method accepted: %d", response.Code)
+	}
+
+	disabledMux := http.NewServeMux()
+	registerStatusHandlers(disabledMux, &healthState{}, false, server, nil, nil)
+
+	disabled := httptest.NewRecorder()
+	disabledMux.ServeHTTP(disabled, httptest.NewRequest(http.MethodPost, path, strings.NewReader("{}")))
+
+	if disabled.Code != http.StatusNotFound {
+		t.Fatalf("aggregated route exposed when disabled: %d", disabled.Code)
+	}
+}

cmd/unbounded-net-controller/detail_api.go

@@ -0,0 +1,132 @@
+// Copyright (c) Microsoft Corporation.
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+
+	"k8s.io/klog/v2"
+
+	"github.com/Azure/unbounded/internal/net/authn"
+	statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1"
+	webhookpkg "github.com/Azure/unbounded/internal/net/webhook"
+)
+
+func registerNodeDetailHandlers(mux *http.ServeMux, health *healthState, requireAuth bool, webhookServer *webhookpkg.Server, authorizer *dashboardAuthorizer, issuer *authn.TokenIssuer) {
+	handler := func(w http.ResponseWriter, r *http.Request) {
+		if !authorizeDashboardOrAggregated(requireAuth, issuer, authorizer, webhookServer, r) {
+			http.Error(w, "Unauthorized", http.StatusUnauthorized)
+
+			return
+		}
+
+		nodeName := r.PathValue("name")
+		manager := health.getDetailRequests()
+
+		if !health.isLeader.Load() || manager == nil {
+			writeNodeDetailResult(w, http.StatusServiceUnavailable,
+				detailRequestFailure(nodeName, r.URL.Query().Get("requestId"), statusv1alpha1.NodeDetailRetryable, "detail request leader is unavailable"))
+
+			return
+		}
+
+		var result statusv1alpha1.NodeDetailResult
+
+		switch r.Method {
+		case http.MethodPost:
+			var input *struct {
+				ForceRefresh bool `json:"forceRefresh"`
+			}
+
+			r.Body = http.MaxBytesReader(w, r.Body, 1<<20)
+			decoder := json.NewDecoder(r.Body)
+			decoder.DisallowUnknownFields()
+
+			err := decoder.Decode(&input)
+			if err == nil {
+				var extra any
+
+				err = decoder.Decode(&extra)
+				if errors.Is(err, io.EOF) && input != nil {
+					err = nil
+				} else if err == nil || input == nil {
+					err = errors.New("expected one JSON object")
+				}
+			}
+
+			if err != nil {
+				code := http.StatusBadRequest
+
+				var tooLarge *http.MaxBytesError
+				if errors.As(err, &tooLarge) {
+					code = http.StatusRequestEntityTooLarge
+				}
+
+				writeNodeDetailResult(w, code, detailRequestFailure(nodeName, "", statusv1alpha1.NodeDetailUnavailable, err.Error()))
+
+				return
+			}
+
+			result = manager.Request(nodeName, input.ForceRefresh)
+		case http.MethodGet:
+			requestID := r.URL.Query().Get("requestId")
+			if requestID == "" {
+				writeNodeDetailResult(w, http.StatusBadRequest,
+					detailRequestFailure(nodeName, "", statusv1alpha1.NodeDetailUnavailable, "requestId is required"))
+
+				return
+			}
+
+			result = manager.Result(nodeName, requestID)
+		default:
+			w.Header().Set("Allow", "GET, POST")
+			http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+
+			return
+		}
+
+		writeNodeDetailResult(w, nodeDetailHTTPStatus(result.State), result)
+	}
+	mux.HandleFunc("/status/node/{name}/details", handler)
+
+	if health.registerAggregatedAPIServer {
+		mux.HandleFunc("/apis/status.net.unbounded-cloud.io/v1alpha1/nodes/{name}/details", func(w http.ResponseWriter, r *http.Request) {
+			if !webhookServer.IsTrustedAggregatedRequest(r) {
+				http.Error(w, "Forbidden", http.StatusForbidden)
+
+				return
+			}
+
+			handler(w, r)
+		})
+	}
+}
+
+func nodeDetailHTTPStatus(state statusv1alpha1.NodeDetailState) int {
+	switch state {
+	case statusv1alpha1.NodeDetailPending:
+		return http.StatusAccepted
+	case statusv1alpha1.NodeDetailComplete:
+		return http.StatusOK
+	case statusv1alpha1.NodeDetailExpired:
+		return http.StatusGone
+	case statusv1alpha1.NodeDetailUnavailable:
+		return http.StatusNotFound
+	default:
+		return http.StatusServiceUnavailable
+	}
+}
+
+func writeNodeDetailResult(w http.ResponseWriter, code int, result statusv1alpha1.NodeDetailResult) {
+	w.Header().Set("Content-Type", "application/json")
+	w.Header().Set("Cache-Control", "no-store")
+	w.WriteHeader(code)
+
+	if err := json.NewEncoder(w).Encode(result); err != nil {
+		klog.V(4).Infof("node detail response encode failed: %v", err)
+	}
+}