feat(observability-proxy): rebuild logs queries from an allowlist

vlselect has no notion of a tenant: whatever filter reaches it decides
what is returned. The upstream parameters are therefore rebuilt from a
per-endpoint allowlist with the namespace scope as the only
extra_stream_filters value, so a client cannot widen it through
extra_filters, tenant headers or an alternate parameter spelling.

VictoriaLogs ANDs extra_stream_filters into every subquery, so union
and in() queries pass through unparsed. What the query can still do is
cost: a missing start scans the whole retention, a tiny step returns a
bucket per millisecond, query has no default limit, and
ignore_global_time_filter lifts the window altogether. Each of those is
bounded or refused here, and a tail resume point becomes a capped
start_offset.

Assisted-by: LLM
Signed-off-by: ZverGuy <maximbel2003@gmail.com>
This commit is contained in:
ZverGuy committed 2026-11-09 15:48:57 +07:00
1 parent e02d0fa16e
commit eb0d6e5957
2 files changed
+586

No files matched your search

+314
View File
@@ -0,0 +1,314 @@
/*
Copyright 2026 The Cozystack Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package observabilityproxy
import (
"fmt"
"net/http"
"net/url"
"regexp"
"strconv"
"strings"
"time"
)
// LogsLimits bound what one logs request may make vlselect do.
type LogsLimits struct {
DefaultRange time.Duration
MaxRange time.Duration
MaxBuckets int
DefaultLimit int
MaxLimit int
MaxTailResume time.Duration
}
// tailOffset is vlselect's default tail offset. The proxy never forwards a
// client offset, so the resume arithmetic can rely on this value.
const tailOffset = 5 * time.Second
const maxHitsFields = 10
var logsEndpoints = map[string]bool{
"query": true, "hits": true, "stats_query_range": true, "tail": true,
"field_values": true, "stream_field_values": true, "facets": true,
}
// BuildLogsParams rebuilds the upstream parameters from scratch: only the
// allowlisted parameters of the endpoint are read from the client, and the
// namespace scope is added as the only extra_stream_filters value.
// VictoriaLogs ANDs extra_stream_filters into the query and every subquery
// (union, join, in()), which is what lets the query itself pass through
// unparsed.
func BuildLogsParams(endpoint string, in url.Values, namespace string, now time.Time, lim LogsLimits) (url.Values, *HTTPError) {
if !logsEndpoints[endpoint] {
return nil, errorf(http.StatusNotFound, "not found")
}
query := in.Get("query")
if query == "" {
query = "*"
}
// The option lifts the time window this function enforces, and on tail
// it rescans the whole retention every tick.
if strings.Contains(strings.ToLower(query), "ignore_global_time_filter") {
return nil, errorf(http.StatusBadRequest, "ignore_global_time_filter is not allowed")
}
out := url.Values{}
out.Set("query", query)
out.Set("extra_stream_filters", "{kubernetes_namespace_name="+strconv.Quote(namespace)+"}")
if endpoint == "tail" {
return out, tailParams(in, out, now, lim)
}
start, end, herr := window(in, now, lim)
if herr != nil {
return nil, herr
}
out.Set("start", start.Format(time.RFC3339Nano))
out.Set("end", end.Format(time.RFC3339Nano))
switch endpoint {
case "query":
herr = setLimit(out, in, "limit", lim.DefaultLimit, lim.MaxLimit)
case "hits":
if herr = setStep(out, in, end.Sub(start), lim); herr != nil {
return nil, herr
}
fields := in["field"]
if len(fields) > maxHitsFields {
return nil, errorf(http.StatusBadRequest, "at most %d field parameters are allowed", maxHitsFields)
}
for _, f := range fields {
if f != "" {
out.Add("field", f)
}
}
if in.Has("fields_limit") {
herr = setLimit(out, in, "fields_limit", 0, lim.MaxLimit)
}
case "stats_query_range":
herr = setStep(out, in, end.Sub(start), lim)
case "field_values", "stream_field_values":
if in.Get("field") == "" {
return nil, errorf(http.StatusBadRequest, "field is required")
}
out.Set("field", in.Get("field"))
if f := in.Get("filter"); f != "" {
out.Set("filter", f)
}
herr = setLimit(out, in, "limit", lim.DefaultLimit, lim.MaxLimit)
case "facets":
for _, k := range []string{"limit", "max_values_per_field", "max_value_len"} {
if k == "limit" || in.Has(k) {
if herr = setLimit(out, in, k, lim.DefaultLimit, lim.MaxLimit); herr != nil {
return nil, herr
}
}
}
if v := in.Get("keep_const_fields"); v == "1" || v == "true" {
out.Set("keep_const_fields", "1")
}
}
if herr != nil {
return nil, herr
}
return out, nil
}
// window derives a finite [start, end]: VictoriaLogs otherwise starts at
// the oldest stored entry and scans the whole retention.
func window(in url.Values, now time.Time, lim LogsLimits) (time.Time, time.Time, *HTTPError) {
end := now
if v := in.Get("end"); v != "" {
t, err := parseTime(v, now)
if err != nil {
return time.Time{}, time.Time{}, errorf(http.StatusBadRequest, "invalid end: %v", err)
}
end = t
}
start := end.Add(-lim.DefaultRange)
if v := in.Get("start"); v != "" {
t, err := parseTime(v, now)
if err != nil {
return time.Time{}, time.Time{}, errorf(http.StatusBadRequest, "invalid start: %v", err)
}
start = t
}
if start.After(end) {
return time.Time{}, time.Time{}, errorf(http.StatusBadRequest, "start is after end")
}
if end.Sub(start) > lim.MaxRange {
start = end.Add(-lim.MaxRange)
}
return start.UTC(), end.UTC(), nil
}
// setStep raises step so that hits and stats_query_range, which return one
// bucket per step, return at most MaxBuckets buckets over the window.
func setStep(out, in url.Values, span time.Duration, lim LogsLimits) *HTTPError {
minStep := max((span+time.Duration(lim.MaxBuckets)-1)/time.Duration(lim.MaxBuckets), time.Second)
step := minStep
if v := in.Get("step"); v != "" {
d, err := parseDuration(v)
if err != nil || d <= 0 {
return errorf(http.StatusBadRequest, "invalid step %q", v)
}
step = max(d, minStep)
}
out.Set("step", formatDuration(step))
return nil
}
func setLimit(out, in url.Values, key string, def, maxv int) *HTTPError {
n := def
if v := in.Get(key); v != "" {
parsed, err := strconv.Atoi(v)
if err != nil || parsed <= 0 {
return errorf(http.StatusBadRequest, "invalid %s %q", key, v)
}
n = parsed
}
out.Set(key, strconv.Itoa(min(n, maxv)))
return nil
}
// tailParams turns a resume point into a capped start_offset. A reconnect
// older than the cap resumes at the cap; the gap needs a historical query.
func tailParams(in, out url.Values, now time.Time, lim LogsLimits) *HTTPError {
var start, end time.Time
for _, k := range []string{"start", "end"} {
v := in.Get(k)
if v == "" {
continue
}
t, err := parseTime(v, now)
if err != nil {
return errorf(http.StatusBadRequest, "invalid %s: %v", k, err)
}
out.Set(k, t.UTC().Format(time.RFC3339Nano))
if k == "start" {
start = t
} else {
end = t
}
}
if !start.IsZero() && !end.IsZero() && start.After(end) {
return errorf(http.StatusBadRequest, "start is after end")
}
since, offset := in.Get("since"), in.Get("start_offset")
var d time.Duration
switch {
case since != "" && offset != "":
return errorf(http.StatusBadRequest, "since and start_offset are mutually exclusive")
case since != "":
t, err := parseTime(since, now)
if err != nil {
return errorf(http.StatusBadRequest, "invalid since: %v", err)
}
d = now.Add(-tailOffset).Sub(t)
case offset != "":
parsed, err := parseDuration(offset)
if err != nil {
return errorf(http.StatusBadRequest, "invalid start_offset %q", offset)
}
d = parsed
default:
return nil
}
out.Set("start_offset", formatDuration(min(max(d, 0), lim.MaxTailResume)))
return nil
}
// formatDuration writes whole milliseconds, a form every VictoriaLogs
// duration parser accepts.
func formatDuration(d time.Duration) string {
return fmt.Sprintf("%dms", d.Milliseconds())
}
var durationToken = regexp.MustCompile(`^(\d+(?:\.\d+)?)(ms|s|m|h|d|w|y)`)
var durationUnits = map[string]time.Duration{
"ms": time.Millisecond, "s": time.Second, "m": time.Minute, "h": time.Hour,
"d": 24 * time.Hour, "w": 7 * 24 * time.Hour, "y": 365 * 24 * time.Hour,
}
// parseDuration accepts the LogsQL duration forms (1h30m, 1d, 2w) and bare
// seconds; Go's parser knows neither d nor w.
func parseDuration(s string) (time.Duration, error) {
if f, err := strconv.ParseFloat(s, 64); err == nil {
if f < 0 {
return 0, fmt.Errorf("negative duration %q", s)
}
return time.Duration(f * float64(time.Second)), nil
}
if s == "" {
return 0, fmt.Errorf("empty duration")
}
var total time.Duration
for rest := s; rest != ""; {
m := durationToken.FindStringSubmatch(rest)
if m == nil {
return 0, fmt.Errorf("invalid duration %q", s)
}
f, _ := strconv.ParseFloat(m[1], 64)
total += time.Duration(f * float64(durationUnits[m[2]]))
rest = rest[len(m[0]):]
}
return total, nil
}
// parseTime accepts "now", unix seconds with an optional fraction, RFC3339,
// and a duration meaning that long before now (VictoriaLogs' "start=3h").
// Anything else is refused rather than guessed, since a misread timestamp
// silently moves the window.
func parseTime(s string, now time.Time) (time.Time, error) {
if s == "now" {
return now, nil
}
if sec, frac, ok := strings.Cut(s, "."); isDigits(sec) && (!ok || isDigits(frac)) {
secs, err := strconv.ParseInt(sec, 10, 64)
if err != nil {
return time.Time{}, err
}
var nanos int64
if ok {
frac = (frac + "000000000")[:9]
nanos, _ = strconv.ParseInt(frac, 10, 64)
}
return time.Unix(secs, nanos), nil
}
if t, err := time.Parse(time.RFC3339Nano, s); err == nil {
return t, nil
}
if d, err := parseDuration(s); err == nil {
return now.Add(-d), nil
}
return time.Time{}, fmt.Errorf("unsupported time %q", s)
}
func isDigits(s string) bool {
if s == "" {
return false
}
for _, r := range s {
if r < '0' || r > '9' {
return false
}
}
return true
}
@@ -0,0 +1,272 @@
/*
Copyright 2026 The Cozystack Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package observabilityproxy
import (
"net/http"
"net/url"
"sort"
"strings"
"testing"
"time"
)
var testNow = time.Date(2026, 11, 4, 12, 0, 0, 0, time.UTC)
func testLimits() LogsLimits {
return LogsLimits{
DefaultRange: time.Hour,
MaxRange: 7 * 24 * time.Hour,
MaxBuckets: 1000,
DefaultLimit: 1000,
MaxLimit: 5000,
MaxTailResume: 15 * time.Minute,
}
}
const wantScope = `{kubernetes_namespace_name="tenant-a"}`
func build(t *testing.T, endpoint string, in url.Values) url.Values {
t.Helper()
out, err := BuildLogsParams(endpoint, in, "tenant-a", testNow, testLimits())
if err != nil {
t.Fatalf("BuildLogsParams(%s, %v) error: %v", endpoint, in, err)
}
return out
}
func buildErr(t *testing.T, endpoint string, in url.Values) *HTTPError {
t.Helper()
out, err := BuildLogsParams(endpoint, in, "tenant-a", testNow, testLimits())
if err == nil {
t.Fatalf("BuildLogsParams(%s, %v) = %v, want error", endpoint, in, out)
}
return err
}
func TestBuildLogsParamsAllowlist(t *testing.T) {
allowed := map[string][]string{
"query": {"query", "start", "end", "limit"},
"hits": {"query", "start", "end", "step", "field", "fields_limit"},
"stats_query_range": {"query", "start", "end", "step"},
"field_values": {"query", "start", "end", "field", "limit", "filter"},
"stream_field_values": {"query", "start", "end", "field", "limit", "filter"},
"facets": {"query", "start", "end", "limit", "max_values_per_field", "max_value_len", "keep_const_fields"},
"tail": {"query", "start", "end", "start_offset"},
}
hostile := url.Values{
"extra_filters": {`{kubernetes_namespace_name="tenant-b"}`},
"extra_filters[]": {`kubernetes_namespace_name:~".*"`},
"extra_stream_filters": {`{kubernetes_namespace_name=~".*"}`},
"AccountID": {"7"},
"ProjectID": {"7"},
"tenant": {"tenant-b"},
"offset": {"1h"},
"refresh_interval": {"1ms"},
"ignore_pipes": {"1"},
"format": {"csv"},
"timeout": {"1h"},
"query": {"error"},
"field": {"kubernetes_pod_name"},
"filter": {"api"},
"keep_const_fields": {"1"},
"since": {"2026-11-04T11:59:00Z"},
}
for endpoint, keys := range allowed {
t.Run(endpoint, func(t *testing.T) {
out := build(t, endpoint, hostile)
want := map[string]bool{"extra_stream_filters": true}
for _, k := range keys {
want[k] = true
}
for k := range out {
if !want[k] {
t.Errorf("forwarded non-allowlisted parameter %q", k)
}
}
if got := out["extra_stream_filters"]; len(got) != 1 || got[0] != wantScope {
t.Errorf("extra_stream_filters = %q, want exactly [%s]", got, wantScope)
}
if _, ok := out["extra_filters"]; ok {
t.Error("extra_filters forwarded")
}
})
}
}
func TestBuildLogsParamsRejects(t *testing.T) {
tests := []struct {
name string
endpoint string
in url.Values
code int
}{
{"unknown endpoint", "tenant_ids", url.Values{}, http.StatusNotFound},
{"stats_query is not proxied", "stats_query", url.Values{}, http.StatusNotFound},
{"ignore_global_time_filter", "query", url.Values{"query": {`* options(ignore_global_time_filter=true)`}}, http.StatusBadRequest},
{"ignore_global_time_filter in subquery", "hits", url.Values{"query": {`* | union (options(ignore_global_time_filter=true) error)`}}, http.StatusBadRequest},
{"ignore_global_time_filter on tail", "tail", url.Values{"query": {`options(IGNORE_GLOBAL_TIME_FILTER=true) *`}}, http.StatusBadRequest},
{"inverted range", "query", url.Values{"start": {"2026-11-04T11:00:00Z"}, "end": {"2026-11-04T10:00:00Z"}}, http.StatusBadRequest},
{"inverted range on tail", "tail", url.Values{"start": {"2026-11-04T11:00:00Z"}, "end": {"2026-11-04T10:00:00Z"}}, http.StatusBadRequest},
{"bad start", "query", url.Values{"start": {"yesterday"}}, http.StatusBadRequest},
{"bad limit", "query", url.Values{"limit": {"-1"}}, http.StatusBadRequest},
{"zero limit", "query", url.Values{"limit": {"0"}}, http.StatusBadRequest},
{"bad step", "hits", url.Values{"step": {"0s"}}, http.StatusBadRequest},
{"field required", "field_values", url.Values{}, http.StatusBadRequest},
{"too many hits fields", "hits", url.Values{"field": strings.Fields("a b c d e f g h i j k")}, http.StatusBadRequest},
{"since and start_offset", "tail", url.Values{"since": {"2026-11-04T11:59:00Z"}, "start_offset": {"1m"}}, http.StatusBadRequest},
{"bad since", "tail", url.Values{"since": {"soon"}}, http.StatusBadRequest},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
if err := buildErr(t, tt.endpoint, tt.in); err.Code != tt.code {
t.Fatalf("code = %d (%s), want %d", err.Code, err.Message, tt.code)
}
})
}
}
// VictoriaLogs propagates extra_stream_filters into every subquery
// (docs.victoriametrics.com/victorialogs/querying/), so union and in()
// queries are forwarded verbatim with the scope beside them.
func TestBuildLogsParamsSubqueriesKeepScope(t *testing.T) {
for _, q := range []string{
`* | union ({kubernetes_namespace_name="tenant-b"})`,
`_stream_id:in({kubernetes_namespace_name="tenant-b"} | fields _stream_id)`,
} {
out := build(t, "query", url.Values{"query": {q}})
if out.Get("query") != q || out.Get("extra_stream_filters") != wantScope {
t.Errorf("query %q forwarded as %v", q, out)
}
}
}
func TestBuildLogsParamsWindow(t *testing.T) {
ts := func(s string) string {
v, err := time.Parse(time.RFC3339, s)
if err != nil {
t.Fatal(err)
}
return v.UTC().Format(time.RFC3339Nano)
}
tests := []struct {
name string
in url.Values
wantStart, wantEnd string
}{
{"defaults", url.Values{}, ts("2026-11-04T11:00:00Z"), ts("2026-11-04T12:00:00Z")},
{"start only", url.Values{"start": {"2026-11-04T09:00:00Z"}}, ts("2026-11-04T09:00:00Z"), ts("2026-11-04T12:00:00Z")},
{"end only", url.Values{"end": {"2026-11-03T00:00:00Z"}}, ts("2026-11-02T23:00:00Z"), ts("2026-11-03T00:00:00Z")},
{"unix seconds", url.Values{"start": {"1793786400"}, "end": {"1793788200.5"}}, "2026-11-04T10:00:00Z", "2026-11-04T10:30:00.5Z"},
{"relative", url.Values{"start": {"3h"}, "end": {"now"}}, ts("2026-11-04T09:00:00Z"), ts("2026-11-04T12:00:00Z")},
{"relative days", url.Values{"start": {"1d"}}, ts("2026-11-03T12:00:00Z"), ts("2026-11-04T12:00:00Z")},
{"over-wide clamped", url.Values{"start": {"2026-01-01T00:00:00Z"}}, ts("2026-10-28T12:00:00Z"), ts("2026-11-04T12:00:00Z")},
{"empty range allowed", url.Values{"start": {"2026-11-04T10:00:00Z"}, "end": {"2026-11-04T10:00:00Z"}}, ts("2026-11-04T10:00:00Z"), ts("2026-11-04T10:00:00Z")},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
for _, endpoint := range []string{"query", "hits", "stats_query_range", "facets"} {
out := build(t, endpoint, tt.in)
if out.Get("start") != tt.wantStart || out.Get("end") != tt.wantEnd {
t.Errorf("%s: window = [%s, %s], want [%s, %s]", endpoint, out.Get("start"), out.Get("end"), tt.wantStart, tt.wantEnd)
}
}
})
}
}
func TestBuildLogsParamsStepAndLimit(t *testing.T) {
day := url.Values{"start": {"2026-11-03T12:00:00Z"}}
with := func(base url.Values, k, v string) url.Values {
c := url.Values{}
for key, vals := range base {
c[key] = vals
}
c.Set(k, v)
return c
}
// 24h over at most 1000 buckets is 86.4s.
if got := build(t, "hits", with(day, "step", "1s")).Get("step"); got != "86400ms" {
t.Errorf("tiny step = %s, want clamped to 86400ms", got)
}
if got := build(t, "stats_query_range", with(day, "step", "1h")).Get("step"); got != "3600000ms" {
t.Errorf("1h step = %s", got)
}
if got := build(t, "stats_query_range", with(day, "step", "1w")).Get("step"); got != "604800000ms" {
t.Errorf("1w step = %s", got)
}
if got := build(t, "hits", day).Get("step"); got != "86400ms" {
t.Errorf("omitted step = %s, want the bucket-bounded minimum", got)
}
if got := build(t, "hits", url.Values{"start": {"2026-11-04T11:59:59Z"}}).Get("step"); got != "1000ms" {
t.Errorf("omitted step on a tiny window = %s, want 1000ms floor", got)
}
if got := build(t, "query", url.Values{}).Get("limit"); got != "1000" {
t.Errorf("default limit = %s", got)
}
if got := build(t, "query", url.Values{"limit": {"10"}}).Get("limit"); got != "10" {
t.Errorf("limit = %s", got)
}
for _, endpoint := range []string{"query", "field_values", "facets"} {
in := url.Values{"limit": {"1000000"}, "field": {"f"}}
if got := build(t, endpoint, in).Get("limit"); got != "5000" {
t.Errorf("%s: limit = %s, want clamped to 5000", endpoint, got)
}
}
facets := build(t, "facets", url.Values{"max_values_per_field": {"99999"}, "max_value_len": {"99999"}})
if facets.Get("max_values_per_field") != "5000" || facets.Get("max_value_len") != "5000" {
t.Errorf("facets caps not clamped: %v", facets)
}
hits := build(t, "hits", url.Values{"field": {"b", "a"}, "fields_limit": {"99999"}})
fields := hits["field"]
sort.Strings(fields)
if strings.Join(fields, ",") != "a,b" || hits.Get("fields_limit") != "5000" {
t.Errorf("hits fields = %v", hits)
}
}
func TestBuildLogsParamsTail(t *testing.T) {
tests := []struct {
name string
in url.Values
want string
}{
{"live only", url.Values{}, ""},
// now - 5s tail offset - since = 55s
{"since within cap", url.Values{"since": {"2026-11-04T11:59:00Z"}}, "55000ms"},
{"since older than cap", url.Values{"since": {"2026-11-04T08:00:00Z"}}, "900000ms"},
{"since in the future", url.Values{"since": {"2026-11-04T13:00:00Z"}}, "0ms"},
{"start_offset within cap", url.Values{"start_offset": {"2m"}}, "120000ms"},
{"start_offset over cap", url.Values{"start_offset": {"30d"}}, "900000ms"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
out := build(t, "tail", tt.in)
if got := out.Get("start_offset"); got != tt.want {
t.Errorf("start_offset = %q, want %q", got, tt.want)
}
if out.Has("start") || out.Has("end") || out.Has("limit") {
t.Errorf("tail got a derived window or limit: %v", out)
}
})
}
out := build(t, "tail", url.Values{"start": {"2026-11-04T11:00:00Z"}})
if out.Get("start") != "2026-11-04T11:00:00Z" {
t.Errorf("tail start = %q, want the client value ANDed into each tick", out.Get("start"))
}
}