Extract selectors from logs query to enable more fine-grained access control (observatorium#555)

Author: shwetaap

Diff for: README.md

@@ -91,6 +91,8 @@ Usage of ./observatorium-api:
     	The log format to use. Options: 'logfmt', 'json'. (default "logfmt")
   -log.level string
     	The log filtering level. Options: 'error', 'warn', 'info', 'debug'. (default "info")
+  -logs.auth.extract-selectors string
+    	Comma-separated list of stream selectors that should be extracted from queries and sent to OPA during authorization.
   -logs.read.endpoint string
     	The endpoint against which to make read requests for logs.
   -logs.rules.endpoint string

Diff for: authorization/grpc.go

@@ -67,7 +67,7 @@ func WithGRPCAuthorizers(authorizers map[string]rbac.Authorizer, methReq GRPCRBa
 			return ctx, status.Error(codes.Unauthenticated, "error finding tenant id")
 		}
 
-		_, ok, data := a.Authorize(subject, groups, accessReq.Permission, accessReq.Resource, tenant, tenantID, token)
+		_, ok, data := a.Authorize(subject, groups, accessReq.Permission, accessReq.Resource, tenant, tenantID, token, nil)
 		if !ok {
 			level.Debug(logger).Log("msg", "gRPC Authorizer: insufficient auth", "subject", subject, "tenant", tenant)
 			return ctx, status.Error(codes.PermissionDenied, "forbidden")

Diff for: authorization/http.go

@@ -2,6 +2,7 @@ package authorization
 
 import (
 	"context"
+	"fmt"
 	"net/http"
 
 	"github.com/observatorium/api/authentication"
@@ -16,6 +17,20 @@ const (
 	// authorizationDataKey is the key that holds the authorization response data
 	// in a request context.
 	authorizationDataKey contextKey = "authzData"
+
+	// authorizationSelectorsKey is the key that holds the data about selectors present in the query.
+	authorizationSelectorsKey contextKey = "authzQuerySelectors"
+)
+
+type SelectorsInfo struct {
+	Selectors   map[string][]string
+	HasWildcard bool
+}
+
+var (
+	emptySelectorsInfo = &SelectorsInfo{
+		Selectors: map[string][]string{},
+	}
 )
 
 // GetData extracts the authz response data from provided context.
@@ -31,6 +46,49 @@ func WithData(ctx context.Context, data string) context.Context {
 	return context.WithValue(ctx, authorizationDataKey, data)
 }
 
+// GetSelectorsInfo extracts the query namespaces from the provided context.
+func GetSelectorsInfo(ctx context.Context) (*SelectorsInfo, bool) {
+	value := ctx.Value(authorizationSelectorsKey)
+	namespaces, ok := value.(*SelectorsInfo)
+
+	return namespaces, ok
+}
+
+// WithSelectorsInfo extends the provided context with the query namespaces.
+func WithSelectorsInfo(ctx context.Context, info *SelectorsInfo) context.Context {
+	return context.WithValue(ctx, authorizationSelectorsKey, info)
+}
+
+// WithLogsStreamSelectorsExtractor returns a middleware that, when enabled, tries to extract
+// stream selectors from queries, so that they can be used in authorizing the request.
+func WithLogsStreamSelectorsExtractor(selectorNames []string) func(http.Handler) http.Handler {
+	enabled := len(selectorNames) > 0
+
+	selectorNameMap := make(map[string]bool, len(selectorNames))
+	for _, l := range selectorNames {
+		selectorNameMap[l] = true
+	}
+
+	return func(next http.Handler) http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if !enabled {
+				next.ServeHTTP(w, r)
+
+				return
+			}
+
+			selectorsInfo, err := extractLogStreamSelectors(selectorNameMap, r.URL.Query())
+			if err != nil {
+				httperr.PrometheusAPIError(w, fmt.Sprintf("error extracting selectors from query: %s", err), http.StatusInternalServerError)
+
+				return
+			}
+
+			next.ServeHTTP(w, r.WithContext(WithSelectorsInfo(r.Context(), selectorsInfo)))
+		})
+	}
+}
+
 // WithAuthorizers returns a middleware that authorizes subjects taken from a request context
 // for the given permission on the given resource for a tenant taken from a request context.
 func WithAuthorizers(authorizers map[string]rbac.Authorizer, permission rbac.Permission, resource string) func(http.Handler) http.Handler {
@@ -74,7 +132,20 @@ func WithAuthorizers(authorizers map[string]rbac.Authorizer, permission rbac.Per
 				return
 			}
 
-			statusCode, ok, data := a.Authorize(subject, groups, permission, resource, tenant, tenantID, token)
+			selectorsInfo, ok := GetSelectorsInfo(r.Context())
+			if !ok {
+				selectorsInfo = emptySelectorsInfo
+			}
+
+			metadataOnly := isMetadataRequest(r.URL.Path)
+
+			extraAttributes := &rbac.ExtraAttributes{
+				Selectors:         selectorsInfo.Selectors,
+				WildcardSelectors: selectorsInfo.HasWildcard,
+				MetadataOnly:      metadataOnly,
+			}
+
+			statusCode, ok, data := a.Authorize(subject, groups, permission, resource, tenant, tenantID, token, extraAttributes)
 			if !ok {
 				// Send 403 http.StatusForbidden
 				w.WriteHeader(statusCode)

Diff for: authorization/meta.go

@@ -0,0 +1,32 @@
+package authorization
+
+import (
+	"strings"
+)
+
+var (
+	metaAbsolutePaths = map[string]bool{
+		"/loki/api/v1/label":  true,
+		"/loki/api/v1/labels": true,
+		"/loki/api/v1/series": true,
+		"/api/prom/label":     true,
+		"/api/prom/series":    true,
+	}
+
+	metaPathLabelValuesNewPrefix = "/loki/api/v1/label/"
+	metaPathLabelValuesOldPrefix = "/api/prom/label/"
+	metaPathLabelValuesSuffix    = "/values"
+)
+
+func isMetadataRequest(path string) bool {
+	if absolutePath := metaAbsolutePaths[path]; absolutePath {
+		return true
+	}
+
+	if (strings.HasPrefix(path, metaPathLabelValuesOldPrefix) || strings.HasPrefix(path, metaPathLabelValuesNewPrefix)) &&
+		strings.HasSuffix(path, metaPathLabelValuesSuffix) {
+		return true
+	}
+
+	return false
+}

Diff for: authorization/meta_test.go

@@ -0,0 +1,30 @@
+package authorization
+
+import "testing"
+
+func TestIsMetaRequest(t *testing.T) {
+	tests := []struct {
+		path string
+		want bool
+	}{
+		{
+			path: "/loki/api/v1/labels",
+			want: true,
+		},
+		{
+			path: "/loki/api/v1/label/kubernetes_namespace_name/values",
+			want: true,
+		},
+		{
+			path: "/loki/api/v1/query_range",
+			want: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.path, func(t *testing.T) {
+			if got := isMetadataRequest(tt.path); got != tt.want {
+				t.Errorf("isMetaRequest() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

Diff for: authorization/query.go

@@ -0,0 +1,75 @@
+package authorization
+
+import (
+	"net/url"
+	"strings"
+
+	logqlv2 "github.com/observatorium/api/logql/v2"
+	"github.com/prometheus/prometheus/model/labels"
+)
+
+func extractLogStreamSelectors(selectorNames map[string]bool, values url.Values) (*SelectorsInfo, error) {
+	query := values.Get("query")
+	if query == "" {
+		return emptySelectorsInfo, nil
+	}
+
+	selectors, hasWildcard, err := parseLogStreamSelectors(selectorNames, query)
+	if err != nil {
+		return nil, err
+	}
+
+	return &SelectorsInfo{
+		Selectors:   selectors,
+		HasWildcard: hasWildcard,
+	}, nil
+}
+
+func parseLogStreamSelectors(selectorNames map[string]bool, query string) (map[string][]string, bool, error) {
+	expr, err := logqlv2.ParseExpr(query)
+	if err != nil {
+		return nil, false, err
+	}
+
+	selectors := make(map[string][]string)
+	appendSelector := func(selector, value string) {
+		values, ok := selectors[selector]
+		if !ok {
+			values = make([]string, 0)
+		}
+
+		values = append(values, value)
+		selectors[selector] = values
+	}
+
+	hasWildcard := false
+	expr.Walk(func(expr interface{}) {
+		switch le := expr.(type) {
+		case *logqlv2.StreamMatcherExpr:
+			for _, m := range le.Matchers() {
+				if _, ok := selectorNames[m.Name]; !ok {
+					continue
+				}
+
+				switch m.Type {
+				case labels.MatchEqual:
+					appendSelector(m.Name, m.Value)
+				case labels.MatchRegexp:
+					values := strings.Split(m.Value, "|")
+					for _, v := range values {
+						if strings.ContainsAny(v, ".+*") {
+							hasWildcard = true
+							continue
+						}
+
+						appendSelector(m.Name, v)
+					}
+				}
+			}
+		default:
+			// Do nothing
+		}
+	})
+
+	return selectors, hasWildcard, nil
+}

Diff for: authorization/query_test.go

@@ -0,0 +1,65 @@
+package authorization
+
+import (
+	"reflect"
+	"testing"
+)
+
+func Test_parseQuerySelectors(t *testing.T) {
+	testSelectorLabels := map[string]bool{
+		"namespace":             true,
+		"other_namespace_label": true,
+	}
+	tests := []struct {
+		query           string
+		wantSelectors   map[string][]string
+		wantHasWildcard bool
+	}{
+		{
+			query: `{namespace="test"}`,
+			wantSelectors: map[string][]string{
+				"namespace": {"test"},
+			},
+		},
+		{
+			query: `{namespace="test",other_namespace_label="test2"}`,
+			wantSelectors: map[string][]string{
+				"namespace":             {"test"},
+				"other_namespace_label": {"test2"},
+			},
+		},
+		{
+			query: `{namespace="test",namespace="test2"}`,
+			wantSelectors: map[string][]string{
+				"namespace": {"test", "test2"},
+			},
+		},
+		{
+			query: `{namespace=~"test|test2"}`,
+			wantSelectors: map[string][]string{
+				"namespace": {"test", "test2"},
+			},
+		},
+		{
+			query: `{namespace=~"test|test2|test3.+"}`,
+			wantSelectors: map[string][]string{
+				"namespace": {"test", "test2"},
+			},
+			wantHasWildcard: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.query, func(t *testing.T) {
+			gotNamespaces, gotHasWildcard, err := parseLogStreamSelectors(testSelectorLabels, tt.query)
+			if err != nil {
+				t.Errorf("parseLogStreamSelectors() error = %v", err)
+			}
+			if !reflect.DeepEqual(gotNamespaces, tt.wantSelectors) {
+				t.Errorf("parseLogStreamSelectors() got = %v, want %v", gotNamespaces, tt.wantSelectors)
+			}
+			if gotHasWildcard != tt.wantHasWildcard {
+				t.Errorf("parseLogStreamSelectors() got = %v, want %v", gotHasWildcard, tt.wantHasWildcard)
+			}
+		})
+	}
+}

Diff for: main.go

@@ -159,8 +159,9 @@ type logsConfig struct {
 	tenantHeader         string
 	tenantLabel          string
 	// Allow only read-only access on rules
-	rulesReadOnly     bool
-	rulesLabelFilters map[string][]string
+	rulesReadOnly        bool
+	rulesLabelFilters    map[string][]string
+	authExtractSelectors []string
 	// enable logs at least one {read,write,tail}Endpoint} is provided.
 	enabled bool
 }
@@ -714,6 +715,7 @@ func main() {
 								logsv1.WithWriteMiddleware(writePathRedirectProtection),
 								logsv1.WithGlobalMiddleware(authentication.WithTenantMiddlewares(pm.Middlewares)),
 								logsv1.WithGlobalMiddleware(authentication.WithTenantHeader(cfg.logs.tenantHeader, tenantIDs)),
+								logsv1.WithReadMiddleware(authorization.WithLogsStreamSelectorsExtractor(cfg.logs.authExtractSelectors)),
 								logsv1.WithReadMiddleware(authorization.WithAuthorizers(authorizers, rbac.Read, "logs")),
 								logsv1.WithReadMiddleware(logsv1.WithEnforceAuthorizationLabels()),
 								logsv1.WithWriteMiddleware(authorization.WithAuthorizers(authorizers, rbac.Write, "logs")),
@@ -984,6 +986,7 @@ func parseFlags() (config, error) {
 		rawLogsTailEndpoint            string
 		rawLogsWriteEndpoint           string
 		rawLogsRuleLabelFilters        string
+		rawLogsAuthExtractSelectors    string
 		rawTracesReadEndpoint          string
 		rawTracesWriteEndpoint         string
 		rawTracingEndpointType         string
@@ -1048,6 +1051,8 @@ func parseFlags() (config, error) {
 		"The name of the rules label that should hold the tenant ID in logs upstreams.")
 	flag.StringVar(&rawLogsWriteEndpoint, "logs.write.endpoint", "",
 		"The endpoint against which to make write requests for logs.")
+	flag.StringVar(&rawLogsAuthExtractSelectors, "logs.auth.extract-selectors", "",
+		"Comma-separated list of stream selectors that should be extracted from queries and sent to OPA during authorization.")
 	flag.StringVar(&rawMetricsReadEndpoint, "metrics.read.endpoint", "",
 		"The endpoint against which to send read requests for metrics. It used as a fallback to 'query.endpoint' and 'query-range.endpoint'.")
 	flag.StringVar(&rawMetricsWriteEndpoint, "metrics.write.endpoint", "",
@@ -1176,6 +1181,10 @@ func parseFlags() (config, error) {
 		}
 
 		cfg.logs.readEndpoint = logsReadEndpoint
+
+		if rawLogsAuthExtractSelectors != "" {
+			cfg.logs.authExtractSelectors = strings.Split(rawLogsAuthExtractSelectors, ",")
+		}
 	}
 
 	if rawLogsRulesEndpoint != "" {