From c4f7a53e2a2ab39030ff19ad3d4245d73c01bbe3 Mon Sep 17 00:00:00 2001 From: bro-adm Date: Tue, 11 Aug 2026 13:41:18 +0300 Subject: [PATCH 1/4] [+] made subject in ctx for tests as helper fro the authentication space --- authentication/testing.go | 14 ++++++++++++++ 1 file changed, 14 insertions(+) create mode 100644 authentication/testing.go diff --git a/authentication/testing.go b/authentication/testing.go new file mode 100644 index 000000000..4896dad4d --- /dev/null +++ b/authentication/testing.go @@ -0,0 +1,14 @@ +package authentication + +import "context" + +// WithSubjectForTesting adds a subject to the context for testing purposes only. +// This should only be used in tests outside the authentication package that need to +// simulate an authenticated request (e.g., testing middleware that depends on GetSubject). +// +// Note: This function exists because subjectKey is intentionally private to prevent +// direct manipulation in production code. In real scenarios, subjects are set by +// authentication middleware (OIDC, mTLS, OpenShift, etc.). +func WithSubjectForTesting(ctx context.Context, subject string) context.Context { + return context.WithValue(ctx, subjectKey, subject) +} From 967ebe8c41fa0f07a4d6e5271f9a6b2265b174da Mon Sep 17 00:00:00 2001 From: bro-adm Date: Tue, 11 Aug 2026 13:41:39 +0300 Subject: [PATCH 2/4] [+] made an admin query detector based on existsence of user id field in query --- api/logs/v1/admin_query_detector.go | 98 +++++++++ api/logs/v1/admin_query_detector_test.go | 260 +++++++++++++++++++++++ 2 files changed, 358 insertions(+) create mode 100644 api/logs/v1/admin_query_detector.go create mode 100644 api/logs/v1/admin_query_detector_test.go diff --git a/api/logs/v1/admin_query_detector.go b/api/logs/v1/admin_query_detector.go new file mode 100644 index 000000000..615956131 --- /dev/null +++ b/api/logs/v1/admin_query_detector.go @@ -0,0 +1,98 @@ +package http + +import ( + "fmt" + "net/http" + "regexp" + "strings" + + "github.com/observatorium/api/authentication" + "github.com/observatorium/api/authorization" + logqlv2 "github.com/observatorium/api/logql/v2" +) + +// WithAdminQueryDetector returns a middleware that detects if a query is an "admin query". +// An admin query is one that: +// - Does not contain the specified user field filter +// - Contains the user field filter with a value different from the authenticated subject (including empty) +// +// When an admin query is detected, a flag is stored in the request context that can be +// used by downstream middleware (like authorization) to modify the resource desired by subject sent to OPA. +func WithAdminQueryDetector(fieldName string) func(http.Handler) http.Handler { + return func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Skip detection for /series endpoint as it uses "match" parameter + // which only supports stream labels, not structured metadata fields + if strings.HasSuffix(r.URL.Path, "/series") { + next.ServeHTTP(w, r) + return + } + + // No query parameter, treat as regular query (not admin) + queryString := r.URL.Query().Get(queryParam) + if queryString == "" { + next.ServeHTTP(w, r) + return + } + + // No subject in context, can't determine admin status + // Let the request continue without setting the flag + subject, ok := authentication.GetSubject(r.Context()) + if !ok { + next.ServeHTTP(w, r) + return + } + + isAdmin, _ := detectAdminQuery(fieldName, subject, queryString) + // Note: We ignore the error from parsing. If parsing fails, detectAdminQuery + // returns true (admin query) as a fail-safe approach. + + // Store the result in context + ctx := authorization.WithIsAdminQuery(r.Context(), isAdmin) + next.ServeHTTP(w, r.WithContext(ctx)) + }) + } +} + +// detectAdminQuery analyzes a LogQL query to determine if it's an admin query. +// Returns true if: +// - The user field is not present in the query +// - The user field value doesn't match the authenticated subject +// Returns false if the user field is present and matches the subject exactly. +func detectAdminQuery(fieldName, subject, queryString string) (bool, error) { + // If parsing fails, treat as admin query (fail-safe) + expr, err := logqlv2.ParseExpr(queryString) + if err != nil { + return true, err + } + + // Walk the AST to find the user field + var foundUserField bool + var userFieldValue string + + expr.Walk(func(e interface{}) { + if logQuery, ok := e.(*logqlv2.LogQueryExpr); ok { + queryStr := logQuery.String() + // Use regex to extract user field value + // Pattern: | fieldName = "value" + pattern := fmt.Sprintf(`\|\s*%s\s*=\s*"([^"]*)"`, regexp.QuoteMeta(fieldName)) + re := regexp.MustCompile(pattern) + matches := re.FindStringSubmatch(queryStr) + if len(matches) >= 2 { + foundUserField = true + userFieldValue = matches[1] + } + } + }) + + // No user field → admin query + if !foundUserField { + return true, nil + } + // Different user or empty value → admin quer + if userFieldValue != subject { + return true, nil + } + + return false, nil // User querying their own logs → NOT admin query +} diff --git a/api/logs/v1/admin_query_detector_test.go b/api/logs/v1/admin_query_detector_test.go new file mode 100644 index 000000000..9d8bf9f6f --- /dev/null +++ b/api/logs/v1/admin_query_detector_test.go @@ -0,0 +1,260 @@ +package http + +import ( + "net/http" + "net/http/httptest" + "net/url" + "testing" + + "github.com/efficientgo/core/testutil" + + "github.com/observatorium/api/authentication" + "github.com/observatorium/api/authorization" +) + +func TestDetectAdminQuery(t *testing.T) { + tt := []struct { + desc string + fieldName string + subject string + queryString string + expectAdmin bool + expectError bool + }{ + { + desc: "query without user field", + fieldName: "user_id", + subject: "john.doe@example.com", + queryString: `{namespace="prod"}`, + expectAdmin: true, + expectError: false, + }, + { + desc: "query with user field matching subject", + fieldName: "user_id", + subject: "john.doe@example.com", + queryString: `{namespace="prod"} | user_id = "john.doe@example.com"`, + expectAdmin: false, + expectError: false, + }, + { + desc: "query with user field NOT matching subject", + fieldName: "user_id", + subject: "john.doe@example.com", + queryString: `{namespace="prod"} | user_id = "jane.smith@example.com"`, + expectAdmin: true, + expectError: false, + }, + { + desc: "query with empty user field value", + fieldName: "user_id", + subject: "john.doe@example.com", + queryString: `{namespace="prod"} | user_id = ""`, + expectAdmin: true, + expectError: false, + }, + { + desc: "query with user field and other filters", + fieldName: "user_id", + subject: "admin@example.com", + queryString: `{namespace="prod"} | user_id = "admin@example.com" | json | level = "error"`, + expectAdmin: false, + expectError: false, + }, + { + desc: "custom field name", + fieldName: "subject", + subject: "user123", + queryString: `{namespace="prod"} | subject = "user123"`, + expectAdmin: false, + expectError: false, + }, + { + desc: "custom field name with mismatch", + fieldName: "subject", + subject: "user123", + queryString: `{namespace="prod"} | subject = "user456"`, + expectAdmin: true, + expectError: false, + }, + { + desc: "invalid LogQL query", + fieldName: "user_id", + subject: "john.doe@example.com", + queryString: `{this is not valid logql`, + expectAdmin: true, + expectError: true, + }, + { + desc: "metric expression without log query", + fieldName: "user_id", + subject: "john.doe@example.com", + queryString: `100 * 100`, + expectAdmin: true, + expectError: false, + }, + { + desc: "query with multiple pipeline stages", + fieldName: "user_id", + subject: "test@example.com", + queryString: `{app="myapp"} | json | line_format "{{.msg}}" | user_id = "test@example.com"`, + expectAdmin: false, + expectError: false, + }, + } + + for _, tc := range tt { + t.Run(tc.desc, func(t *testing.T) { + isAdmin, err := detectAdminQuery(tc.fieldName, tc.subject, tc.queryString) + + if tc.expectError { + testutil.Assert(t, err != nil, "expected error but got none") + } else { + testutil.Ok(t, err) + } + + testutil.Equals(t, tc.expectAdmin, isAdmin, + "expected isAdmin=%v, got isAdmin=%v", tc.expectAdmin, isAdmin) + }) + } +} + +func TestWithAdminQueryDetectorMiddleware(t *testing.T) { + tt := []struct { + desc string + fieldName string + subject string + urlPath string + queryParam string + expectAdminFlag bool + expectFlagPresent bool + }{ + { + desc: "query without user field - should set admin flag", + fieldName: "user_id", + subject: "john.doe@example.com", + urlPath: "/loki/api/v1/query", + queryParam: `{namespace="prod"}`, + expectAdminFlag: true, + expectFlagPresent: true, + }, + { + desc: "query with matching user field - should not set admin flag", + fieldName: "user_id", + subject: "john.doe@example.com", + urlPath: "/loki/api/v1/query", + queryParam: `{namespace="prod"} | user_id = "john.doe@example.com"`, + expectAdminFlag: false, + expectFlagPresent: true, + }, + { + desc: "query with different user field - should set admin flag", + fieldName: "user_id", + subject: "john.doe@example.com", + urlPath: "/loki/api/v1/query", + queryParam: `{namespace="prod"} | user_id = "jane.smith@example.com"`, + expectAdminFlag: true, + expectFlagPresent: true, + }, + { + desc: "series endpoint - should skip detection", + fieldName: "user_id", + subject: "john.doe@example.com", + urlPath: "/loki/api/v1/series", + queryParam: `{namespace="prod"}`, + expectAdminFlag: false, + expectFlagPresent: false, + }, + { + desc: "empty query parameter - should skip detection", + fieldName: "user_id", + subject: "john.doe@example.com", + urlPath: "/loki/api/v1/labels", + queryParam: "", + expectAdminFlag: false, + expectFlagPresent: false, + }, + { + desc: "query_range endpoint with admin query", + fieldName: "user_id", + subject: "admin@example.com", + urlPath: "/loki/api/v1/query_range", + queryParam: `{app="test"}`, + expectAdminFlag: true, + expectFlagPresent: true, + }, + { + desc: "values endpoint with user query", + fieldName: "user_id", + subject: "user@example.com", + urlPath: "/loki/api/v1/label/namespace/values", + queryParam: `{namespace="prod"} | user_id = "user@example.com"`, + expectAdminFlag: false, + expectFlagPresent: true, + }, + } + + for _, tc := range tt { + t.Run(tc.desc, func(t *testing.T) { + var capturedAdminFlag bool + var capturedFlagPresent bool + + // Create a next handler that captures the admin flag from context + nextHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + capturedAdminFlag, capturedFlagPresent = authorization.GetIsAdminQuery(r.Context()) + w.WriteHeader(http.StatusOK) + }) + + // Create the middleware + middleware := WithAdminQueryDetector(tc.fieldName) + handler := middleware(nextHandler) + + // Create request URL with query parameter + reqURL := tc.urlPath + if tc.queryParam != "" { + reqURL += "?query=" + url.QueryEscape(tc.queryParam) + } + req := httptest.NewRequest("GET", reqURL, nil) + + // Add subject to context + ctx := authentication.WithSubjectForTesting(req.Context(), tc.subject) + req = req.WithContext(ctx) + + // Execute the handler + rr := httptest.NewRecorder() + handler.ServeHTTP(rr, req) + + // Verify the flag was set correctly + testutil.Equals(t, tc.expectFlagPresent, capturedFlagPresent, + "expected flag present=%v, got flag present=%v", tc.expectFlagPresent, capturedFlagPresent) + + if tc.expectFlagPresent { + testutil.Equals(t, tc.expectAdminFlag, capturedAdminFlag, + "expected admin flag=%v, got admin flag=%v", tc.expectAdminFlag, capturedAdminFlag) + } + }) + } +} + +func TestWithAdminQueryDetectorMiddleware_NoSubject(t *testing.T) { + // Test that middleware doesn't set flag when subject is missing + var capturedFlagPresent bool + + nextHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, capturedFlagPresent = authorization.GetIsAdminQuery(r.Context()) + w.WriteHeader(http.StatusOK) + }) + + middleware := WithAdminQueryDetector("user_id") + handler := middleware(nextHandler) + + reqURL := "/loki/api/v1/query?query=" + url.QueryEscape(`{namespace="prod"}`) + req := httptest.NewRequest("GET", reqURL, nil) + // Don't add subject to context + + rr := httptest.NewRecorder() + handler.ServeHTTP(rr, req) + + testutil.Equals(t, false, capturedFlagPresent, + "expected no flag when subject is missing") +} From cfc55ea506743c2da9fb1470dd49be536c685052 Mon Sep 17 00:00:00 2001 From: bro-adm Date: Tue, 11 Aug 2026 13:42:42 +0300 Subject: [PATCH 3/4] [+] updated the authorization call to opa to deferiate between logs reource and logs admin resource on case the user field is enabled --- authorization/http.go | 25 ++++++++++++++++++++++++- 1 file changed, 24 insertions(+), 1 deletion(-) diff --git a/authorization/http.go b/authorization/http.go index c59857519..d47b292ce 100644 --- a/authorization/http.go +++ b/authorization/http.go @@ -25,6 +25,10 @@ const ( // authorizationSelectorsKey is the key that holds the data about selectors present in the query. authorizationSelectorsKey contextKey = "authzQuerySelectors" + // isAdminQueryKey is the key that holds the admin query detection flag + // in a request context. + isAdminQueryKey contextKey = "isAdminQuery" + // errorMessageForbidden is the error message presented to the user if the user doesn't have // sufficient permissions to access the requested tenant. errorMessageForbidden string = "You don't have permission to access this tenant" @@ -163,7 +167,15 @@ func WithAuthorizers(authorizers map[string]rbac.Authorizer, permission rbac.Per MetadataOnly: metadataOnly, } - statusCode, ok, data := a.Authorize(subject, groups, permission, resource, tenant, tenantID, token, extraAttributes) + // Check if this is an admin query (from admin query detector middleware) + // This only happens when --logs.user-field flag is set and admin query detector runs + // If flag not set, isAdminQuery won't be in context, and actualResource stays as "logs" + desiredResource := resource + if isAdminQuery, ok := GetIsAdminQuery(r.Context()); ok && isAdminQuery { + desiredResource = resource + "/admin" // "logs" → "logs/admin" + } + + statusCode, ok, data := a.Authorize(subject, groups, permission, desiredResource, tenant, tenantID, token, extraAttributes) if !ok { switch statusCode { case http.StatusForbidden: @@ -179,3 +191,14 @@ func WithAuthorizers(authorizers map[string]rbac.Authorizer, permission rbac.Per }) } } + +// WithIsAdminQuery stores the admin query flag in the context. +func WithIsAdminQuery(ctx context.Context, isAdmin bool) context.Context { + return context.WithValue(ctx, isAdminQueryKey, isAdmin) +} + +// GetIsAdminQuery retrieves the admin query flag from the context. +func GetIsAdminQuery(ctx context.Context) (bool, bool) { + val, ok := ctx.Value(isAdminQueryKey).(bool) + return val, ok +} From e8d2ab115de5aa361cfca5c050058d120675920a Mon Sep 17 00:00:00 2001 From: bro-adm Date: Tue, 11 Aug 2026 13:47:07 +0300 Subject: [PATCH 4/4] [+] conditional is admin mioddleware handler for logs in main piplleine along wiht the user field name in cfg --- README.md | 2 ++ main.go | 51 +++++++++++++++++++++++++++++++++++---------------- 2 files changed, 37 insertions(+), 16 deletions(-) diff --git a/README.md b/README.md index d89f5d90c..b1677d301 100644 --- a/README.md +++ b/README.md @@ -102,6 +102,8 @@ Usage of ./observatorium-api: File containing the TLS client key to authenticate against upstream logs servers. Leave blank to disable mTLS. -logs.tls.watch-certs Watch for certificate changes and reload + -logs.user-field string + The name of the structured metadata field that should hold the user ID in logs queries (e.g., 'user_id'). When not set (default), admin query detection is disabled and all queries use resource='logs'. This is opt-in only to ensure non-breaking changes. -logs.write-timeout duration The HTTP write timeout for proxied requests to the logs endpoint. (default 10m0s) -logs.write.endpoint string diff --git a/main.go b/main.go index af2ff1c78..827631c59 100644 --- a/main.go +++ b/main.go @@ -169,6 +169,7 @@ type logsConfig struct { upstreamKeyFile string tenantHeader string tenantLabel string + userField string // Allow only read-only access on rules rulesReadOnly bool rulesLabelFilters map[string][]string @@ -793,6 +794,37 @@ func main() { r.Group(func(r chi.Router) { r.Use(middleware.Timeout(cfg.logs.upstreamWriteTimeout)) + + handlerOpts := []logsv1.HandlerOption{ + logsv1.Logger(logger), + logsv1.WithRegistry(reg), + logsv1.WithHandlerInstrumenter(instrumenter), + logsv1.WithWriteMiddleware(writePathRedirectProtection), + logsv1.WithGlobalMiddleware(authentication.WithTenantMiddlewares(pm.Middlewares)), + logsv1.WithGlobalMiddleware(authentication.WithTenantHeader(cfg.logs.tenantHeader, tenantIDs)), + logsv1.WithReadMiddleware(authorization.WithLogsStreamSelectorsExtractor(logger, cfg.logs.authExtractSelectors)), + } + + // Conditionally add admin query detector if user field is configured + if cfg.logs.userField != "" { + handlerOpts = append(handlerOpts, + logsv1.WithReadMiddleware(logsv1.WithAdminQueryDetector(cfg.logs.userField)), + ) + } + + // Add remaining middleware + handlerOpts = append(handlerOpts, + logsv1.WithReadMiddleware(authorization.WithAuthorizers(authorizers, rbac.Read, "logs")), + logsv1.WithReadMiddleware(logsv1.WithEnforceAuthorizationLabels()), + logsv1.WithWriteMiddleware(authorization.WithAuthorizers(authorizers, rbac.Write, "logs")), + logsv1.WithRulesLabelFilters(cfg.logs.rulesLabelFilters), + logsv1.WithRulesReadMiddleware(logsv1.WithEnforceTenantAsRuleNamespace()), + logsv1.WithRulesReadMiddleware(logsv1.WithEnforceRulesAuthorizationLabels()), + logsv1.WithRulesReadMiddleware(logsv1.WithParametersAsLabelsFilterRules(cfg.logs.rulesLabelFilters)), + logsv1.WithRulesWriteMiddleware(logsv1.WithEnforceTenantAsRuleNamespace()), + logsv1.WithRulesWriteMiddleware(logsv1.WithEnforceRuleLabels(cfg.logs.tenantLabel)), + ) + r.Mount("/api/logs/v1/{tenant}", stripTenantPrefix("/api/logs/v1", logsv1.NewHandler( @@ -802,22 +834,7 @@ func main() { cfg.logs.rulesEndpoint, cfg.logs.rulesReadOnly, logsUpstreamClientOptions, - logsv1.Logger(logger), - logsv1.WithRegistry(reg), - logsv1.WithHandlerInstrumenter(instrumenter), - logsv1.WithWriteMiddleware(writePathRedirectProtection), - logsv1.WithGlobalMiddleware(authentication.WithTenantMiddlewares(pm.Middlewares)), - logsv1.WithGlobalMiddleware(authentication.WithTenantHeader(cfg.logs.tenantHeader, tenantIDs)), - logsv1.WithReadMiddleware(authorization.WithLogsStreamSelectorsExtractor(logger, cfg.logs.authExtractSelectors)), - logsv1.WithReadMiddleware(authorization.WithAuthorizers(authorizers, rbac.Read, "logs")), - logsv1.WithReadMiddleware(logsv1.WithEnforceAuthorizationLabels()), - logsv1.WithWriteMiddleware(authorization.WithAuthorizers(authorizers, rbac.Write, "logs")), - logsv1.WithRulesLabelFilters(cfg.logs.rulesLabelFilters), - logsv1.WithRulesReadMiddleware(logsv1.WithEnforceTenantAsRuleNamespace()), - logsv1.WithRulesReadMiddleware(logsv1.WithEnforceRulesAuthorizationLabels()), - logsv1.WithRulesReadMiddleware(logsv1.WithParametersAsLabelsFilterRules(cfg.logs.rulesLabelFilters)), - logsv1.WithRulesWriteMiddleware(logsv1.WithEnforceTenantAsRuleNamespace()), - logsv1.WithRulesWriteMiddleware(logsv1.WithEnforceRuleLabels(cfg.logs.tenantLabel)), + handlerOpts..., ), ), ) @@ -1188,6 +1205,8 @@ func parseFlags() (config, error) { "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(&cfg.logs.userField, "logs.user-field", "", + "The name of the structured metadata field that should hold the user ID in logs queries (e.g., 'user_id'). When not set (default), admin query detection is disabled and all queries use resource='logs'. This is opt-in only to ensure non-breaking changes.") flag.StringVar(&rawMetricsReadEndpoint, "metrics.read.endpoint", "", "The endpoint against which to send read requests for metrics.") flag.StringVar(&rawMetricsWriteEndpoint, "metrics.write.endpoint", "",