package server import ( "fmt" "net/url" "strconv" "strings" "time" ) const ( defaultWindow = 15 * time.Minute // maxWindow matches the table TTL; anything wider scans nothing extra // but signals a client bug, so reject it. maxWindow = 72 * time.Hour defaultLimit = 100 maxLimit = 1000 tailLimit = 500 ) // filterColumns maps API query parameters to the column expression they filter // on. Only these names are ever accepted; values are always bound parameters. var filterColumns = []struct { param string column string }{ {"namespace", "namespace"}, {"host", "host"}, {"pod", "pod"}, {"container", "container"}, {"app", "labels['app']"}, {"severity", "severity"}, {"stream", "stream"}, {"source", "source"}, } const selectColumns = "timestamp, toUnixTimestamp64Milli(timestamp) AS ts_ms, host, source, namespace, pod, container, stream, severity, message, labels, fields" type timeRange struct { since time.Time until time.Time } // parseTimeRange enforces server-side time bounds: missing bounds default to a // 15m window ending now, and windows wider than maxWindow are rejected so a // query can never run unbounded over the store. func parseTimeRange(sinceStr, untilStr string, now time.Time) (timeRange, error) { until := now if untilStr != "" { t, err := parseTime(untilStr, now) if err != nil { return timeRange{}, fmt.Errorf("invalid until: %w", err) } until = t } since := until.Add(-defaultWindow) if sinceStr != "" { t, err := parseTime(sinceStr, now) if err != nil { return timeRange{}, fmt.Errorf("invalid since: %w", err) } since = t } if !since.Before(until) { return timeRange{}, fmt.Errorf("since (%s) must be before until (%s)", since.UTC().Format(time.RFC3339), until.UTC().Format(time.RFC3339)) } if until.Sub(since) > maxWindow { return timeRange{}, fmt.Errorf("time window %s exceeds maximum %s", until.Sub(since), maxWindow) } return timeRange{since: since, until: until}, nil } // parseTime accepts RFC3339, "2006-01-02 15:04:05", unix seconds/millis, and // relative durations like "15m" / "6h" / "1d" (meaning that long ago). func parseTime(s string, now time.Time) (time.Time, error) { if t, err := time.Parse(time.RFC3339, s); err == nil { return t, nil } if t, err := time.Parse("2006-01-02 15:04:05", s); err == nil { return t.UTC(), nil } if n, err := strconv.ParseInt(s, 10, 64); err == nil { if n > 1e12 { return time.UnixMilli(n).UTC(), nil } return time.Unix(n, 0).UTC(), nil } if d, err := parseDuration(strings.TrimPrefix(s, "-")); err == nil { return now.Add(-d), nil } return time.Time{}, fmt.Errorf("unrecognised time %q", s) } func parseDuration(s string) (time.Duration, error) { if strings.HasSuffix(s, "d") { n, err := strconv.Atoi(strings.TrimSuffix(s, "d")) if err != nil { return 0, err } return time.Duration(n) * 24 * time.Hour, nil } return time.ParseDuration(s) } func parseLimit(s string, def int) (int, error) { if s == "" { return def, nil } n, err := strconv.Atoi(s) if err != nil || n < 1 { return 0, fmt.Errorf("invalid limit %q", s) } if n > maxLimit { n = maxLimit } return n, nil } func parseOffset(s string) (int, error) { if s == "" { return 0, nil } n, err := strconv.Atoi(s) if err != nil || n < 0 { return 0, fmt.Errorf("invalid offset %q", s) } return n, nil } // validateSQLFragment gates the raw WHERE fragment. Real safety comes from the // readonly ClickHouse user and the parameterized outer query; this only blocks // statement separators. func validateSQLFragment(frag string) error { if strings.Contains(frag, ";") { return fmt.Errorf("sql fragment must not contain ';'") } return nil } // encodeStringArray renders a ClickHouse Array(String) parameter value. func encodeStringArray(items []string) string { var b strings.Builder b.WriteByte('[') for i, it := range items { if i > 0 { b.WriteByte(',') } b.WriteByte('\'') b.WriteString(strings.ReplaceAll(strings.ReplaceAll(it, `\`, `\\`), `'`, `\'`)) b.WriteByte('\'') } b.WriteByte(']') return b.String() } // whereClause builds the parameterized WHERE clause shared by the endpoints. // The time bounds are always present and always bound parameters, so no // filter or raw sql fragment can widen the scanned range. func whereClause(tr timeRange, form url.Values, params map[string]string, exclusiveSince bool) (string, error) { var conds []string params["since_ms"] = strconv.FormatInt(tr.since.UnixMilli(), 10) params["until_ms"] = strconv.FormatInt(tr.until.UnixMilli(), 10) sinceOp := ">=" if exclusiveSince { sinceOp = ">" } conds = append(conds, "timestamp "+sinceOp+" fromUnixTimestamp64Milli({since_ms:Int64})", "timestamp < fromUnixTimestamp64Milli({until_ms:Int64})", ) for _, f := range filterColumns { if v := form.Get(f.param); v != "" { p := "f_" + f.param params[p] = v conds = append(conds, fmt.Sprintf("%s = {%s:String}", f.column, p)) } } if q := strings.TrimSpace(form.Get("q")); q != "" { tokens := strings.Fields(q) params["q_tokens"] = encodeStringArray(tokens) conds = append(conds, "arrayAll(t -> positionCaseInsensitive(message, t) > 0, {q_tokens:Array(String)})") } if frag := strings.TrimSpace(form.Get("sql")); frag != "" { if err := validateSQLFragment(frag); err != nil { return "", err } conds = append(conds, "( "+frag+" )") } return strings.Join(conds, "\n AND "), nil } func buildQuerySQL(where string, limit, offset int) string { return fmt.Sprintf( "SELECT %s\nFROM logs.raw\nWHERE %s\nORDER BY timestamp DESC\nLIMIT %d OFFSET %d\nFORMAT JSON", selectColumns, where, limit, offset) } func buildTailSQL(where string) string { return fmt.Sprintf( "SELECT %s\nFROM logs.raw\nWHERE %s\nORDER BY timestamp ASC\nLIMIT %d\nFORMAT JSON", selectColumns, where, tailLimit) } func buildFacetsSQL(where string) string { facet := func(name, expr string) string { return fmt.Sprintf( "SELECT * FROM (SELECT '%s' AS facet, %s AS value, count() AS n FROM logs.raw WHERE %s GROUP BY value ORDER BY n DESC LIMIT 20)", name, expr, where) } return facet("namespace", "namespace") + "\nUNION ALL\n" + facet("app", "labels['app']") + "\nUNION ALL\n" + facet("host", "host") + "\nFORMAT JSON" }