From c8f45044a3b0185815993a97803701ae80d4fa4c Mon Sep 17 00:00:00 2001 From: Cedric Date: Sun, 1 Nov 2026 15:47:57 +0100 Subject: [PATCH] Reading API /api/v1/logs with a stable cursor, and API tokens GET /api/v1/logs?query=&since=&limit= returns the logs after a cursor, oldest first, with next_cursor and has_more. The cursor is a position on _time (the reception time) plus the number of rows already returned among those sharing that exact time, ordered by a hash of their content, so following the cursors gives every row once even within groups cut by a page. Rows count once older than the lag (30 s), and nothing is returned while the disk spool still holds logs. API tokens (Settings > API) give read-only access to /api/v1/ as "Authorization: Bearer lst_...", in every auth mode; only their SHA-256 is kept in /data/api_tokens.json. README (EN/FR) documents the route and an n8n workflow keeping the cursor in $getWorkflowStaticData. Co-Authored-By: Claude Opus 5.5 (1M context) --- README.fr.md | 82 +++++++++++++- README.md | 76 ++++++++++++- api.go | 5 + apitokens.go | 254 +++++++++++++++++++++++++++++++++++++++++ apitokens_test.go | 176 +++++++++++++++++++++++++++++ logsapi.go | 281 ++++++++++++++++++++++++++++++++++++++++++++++ logsapi_test.go | 276 +++++++++++++++++++++++++++++++++++++++++++++ main.go | 12 +- store.go | 9 ++ web/app.js | 109 +++++++++++++++++- web/index.html | 32 ++++++ web/style.css | 22 ++++ 12 files changed, 1327 insertions(+), 7 deletions(-) create mode 100644 apitokens.go create mode 100644 apitokens_test.go create mode 100644 logsapi.go create mode 100644 logsapi_test.go diff --git a/README.fr.md b/README.fr.md index 72dbb84..8e159e8 100644 --- a/README.fr.md +++ b/README.fr.md @@ -390,6 +390,79 @@ charge (pas de `| pipes`). En ligne de commande : `curl -o logs.csv 'localhost:8080/api/export.csv?q=error&range=24h&tz=Europe/Paris'`. +## API de lecture (n8n, scripts) + +`GET /api/v1/logs` renvoie les logs stockés qui suivent un curseur, du plus ancien au plus +récent : un outil appelé à intervalle régulier (n8n, un script, un connecteur SIEM) reçoit ainsi +chaque nouveau log une fois, sans trou ni doublon, même quand beaucoup de logs ont le même +horodatage. + +L'accès passe par un **jeton d'API**, créé dans *Paramètres > API* par un administrateur et +envoyé dans `Authorization: Bearer lst_…`. Un jeton est en lecture seule et n'ouvre que +`/api/v1/` ; il fonctionne quel que soit le mode d'[authentification](#authentification) +(`local`, `oidc` ou aucune). Il n'est affiché qu'une fois, à sa création : LogStream n'en garde +que l'empreinte SHA-256, dans `/data/api_tokens.json`. Le révoquer coupe l'accès immédiatement ; +la liste indique quand et d'où chaque jeton a servi pour la dernière fois. Une session ouverte +dans le navigateur peut aussi appeler la route. + +| Paramètre | Rôle | +|---|---| +| `query` | filtre LogsQL (par défaut `*`), par exemple `severity:in(err,crit) app:sshd`. Les pipes (`\| ...`) sont refusés. | +| `since` | point de départ : le `next_cursor` de la réponse précédente, une date RFC 3339 (`2026-11-01T08:00:00Z`) ou une durée avant maintenant (`15m`, `6h`, `7d`). Par défaut : `1h`. | +| `limit` | lignes par réponse, de 1 à 10 000 (1 000 par défaut). | +| `lag` | seuls les logs reçus depuis au moins cette durée sont renvoyés (`30s` par défaut, `1h` au plus), pour ne pas sauter les logs encore dans la file ou en attente de leur nom DNS. | + +```json +{ + "rows": [{"_time": "2026-11-01T14:02:11.52Z", "_msg": "…", "host": "fw01", "app": "sshd", "severity": "err", …}], + "next_cursor": "eyJ0IjoiMjAyNi0xMS0wMVQxNDowMjoxMS41MjAwMDAwMDFaIiwicSI6IjJlYjE4YzNhIn0", + "has_more": false, + "until": "2026-11-01T14:05:30.1Z" +} +``` + +Les lignes contiennent les champs stockés (comme la vue JSON de l'interface), triées par +`_time`, l'heure de réception. Renvoyez toujours `next_cursor`, même quand `rows` est vide. +`has_more: true` signifie que d'autres logs sont prêts : rappelez tout de suite. +`waiting: true` signifie que des logs du [tampon disque](#configuration) sont encore en cours +d'envoi à VictoriaLogs : rien n'est renvoyé tant qu'ils ne sont pas stockés, pour n'en manquer +aucun. Le curseur est lié à sa `query` : changer la requête garde la position dans le temps. Les +erreurs arrivent sous la forme `{"error", "code"}` avec le statut 400 (paramètre ou LogsQL +invalide), 401 (jeton invalide ou révoqué) ou 502 (VictoriaLogs injoignable). + +```bash +curl -H "Authorization: Bearer lst_…" \ + "https://logs.example.org/api/v1/logs?query=severity:err&since=15m&limit=1000" +``` + +### Exemple : workflow n8n + +Quatre nœuds, le curseur gardé dans les données statiques du workflow d'une exécution à l'autre : + +1. **Schedule Trigger** : toutes les minutes. +2. **Code** (*Run Once for All Items*), lit le curseur : + ```js + const data = $getWorkflowStaticData('global'); + return [{ json: { since: data.cursor || '15m' } }]; + ``` +3. **HTTP Request** : méthode `GET`, URL `https://logs.example.org/api/v1/logs`. + *Authentication* : Generic Credential Type > Header Auth, avec le nom `Authorization` et la + valeur `Bearer lst_…`. *Send Query Parameters* : `query` = `severity:in(emerg,alert,crit,err)`, + `since` = `{{ $json.since }}`, `limit` = `1000`. +4. **Code** (*Run Once for All Items*), enregistre le curseur et fait de chaque log un item : + ```js + const data = $getWorkflowStaticData('global'); + const page = $input.first().json; + data.cursor = page.next_cursor; + return page.rows.map((row) => ({ json: row })); + ``` + +Branchez ensuite ce que les logs doivent alimenter (Slack, mail, ticket, base de données…). n8n ne +garde les données statiques que pour les exécutions du workflow actif, pas pour les tests +manuels : pendant les essais, chaque exécution repart de `15m`. Au-delà de 1 000 logs par minute, +`has_more` vaut true et les exécutions suivantes rattrapent le retard ; augmentez `limit` ou la +fréquence du déclencheur si besoin. + ## Paramètres L'icône en forme d'engrenage ouvre les paramètres, organisés en onglets. Les modifications @@ -397,8 +470,8 @@ s'affichent tout de suite mais ne sont conservées qu'avec le bouton **Valider** des onglets : il prend la couleur d'accent tant que des modifications sont en attente. **Annuler** revient aux paramètres enregistrés, et fermer la fenêtre avec des modifications non validées demande confirmation avant de les abandonner. Localisation et Affichage sont mémorisés par -navigateur ; les tags de couleur et les sources de logs sont enregistrés sur le serveur. La purge -et la réinitialisation des colonnes agissent immédiatement. +navigateur ; les tags de couleur et les sources de logs sont enregistrés sur le serveur. La purge, +les jetons d'API et la réinitialisation des colonnes agissent immédiatement. - **Localisation et langue** - *Langue* : anglais ou français. @@ -452,6 +525,9 @@ et la réinitialisation des colonnes agissent immédiatement. désactivée par défaut : mettez `ALLOW_PURGE=true` pour l'autoriser. Tout administrateur peut alors purger : activez l'[authentification](#authentification) si l'interface est accessible à d'autres. +- **API** : jetons de l'[API de lecture](#api-de-lecture-n8n-scripts) : en créer un avec un nom + (affiché une seule fois, à copier tout de suite), voir quand et d'où chacun a servi pour la + dernière fois, le révoquer. Réservé aux administrateurs. ## Authentification @@ -631,6 +707,8 @@ Pour mettre à jour l'une d'elles : | `hub.go` | envoie les nouveaux messages aux navigateurs (SSE) | | `rdns.go` | résolutions DNS inverses avec cache | | `export.go` | export CSV en flux | +| `logsapi.go` | API de lecture `/api/v1/logs` : curseur, groupes de logs de même heure | +| `apitokens.go` | jetons d'API : stockage (haché), contrôle Bearer, `/api/tokens` | | `docker.go` | logs des conteneurs Docker (API, lecteurs, positions, détection du niveau) | | `syslogserver.go` | écoutes syslog ouvertes et fermées depuis Paramètres > Sources de logs | | `hostlogs.go`, `journal.go` | logs système de l'hôte : lecteur du journal systemd (sans `journalctl`) et suivi de `/var/log` | diff --git a/README.md b/README.md index 8788f67..75f39ac 100644 --- a/README.md +++ b/README.md @@ -354,13 +354,82 @@ chosen in Settings), `severity`, `facility`, `host`, `host_ip`, `app`, `pid`, `s From the command line: `curl -o logs.csv 'localhost:8080/api/export.csv?q=error&range=24h&tz=Europe/Paris'`. +## Reading API (n8n, scripts) + +`GET /api/v1/logs` returns the stored logs that follow a cursor, oldest first, so that a tool +called on a schedule (n8n, a script, a SIEM connector) gets every new log once, without gaps +or duplicates, even when many logs share the same timestamp. + +Access goes through an **API token**, created in *Settings > API* by an administrator and sent +as `Authorization: Bearer lst_…`. A token is read-only and only opens `/api/v1/`; it works +whatever the [authentication](#authentication) mode (`local`, `oidc`, or none). It is shown once, +when it is created: LogStream only keeps its SHA-256 hash, in `/data/api_tokens.json`. +Revoking it cuts off access at once; the list shows when and from where each token was last +used. A logged-in browser session can call the route too. + +| Parameter | Meaning | +|---|---| +| `query` | LogsQL filter (default `*`), e.g. `severity:in(err,crit) app:sshd`. Pipes (`\| ...`) are refused. | +| `since` | where to start: the `next_cursor` of the previous answer, an RFC 3339 time (`2026-11-01T08:00:00Z`) or a duration back from now (`15m`, `6h`, `7d`). Default: `1h`. | +| `limit` | rows per answer, 1 to 10,000 (default 1,000). | +| `lag` | only logs received at least this long ago are returned (default `30s`, at most `1h`), so that the logs still in the queue or waiting for their DNS name are not skipped. | + +```json +{ + "rows": [{"_time": "2026-11-01T14:02:11.52Z", "_msg": "…", "host": "fw01", "app": "sshd", "severity": "err", …}], + "next_cursor": "eyJ0IjoiMjAyNi0xMS0wMVQxNDowMjoxMS41MjAwMDAwMDFaIiwicSI6IjJlYjE4YzNhIn0", + "has_more": false, + "until": "2026-11-01T14:05:30.1Z" +} +``` + +Rows have the stored fields (as in the UI's JSON view), sorted by `_time`, the reception time. +Always send back `next_cursor`, even when `rows` is empty. `has_more: true` means more logs are +ready: call again at once. `waiting: true` means logs from the [disk buffer](#configuration) are +still being sent to VictoriaLogs: nothing is returned until they are stored, so none is missed. +The cursor belongs to its `query`: changing the query keeps the position in time. Errors come +as `{"error", "code"}` with status 400 (bad parameter or LogsQL), 401 (invalid or revoked token) +or 502 (VictoriaLogs unreachable). + +```bash +curl -H "Authorization: Bearer lst_…" \ + "https://logs.example.org/api/v1/logs?query=severity:err&since=15m&limit=1000" +``` + +### Example: n8n workflow + +Four nodes, the cursor kept in the workflow's static data between runs: + +1. **Schedule Trigger**: every minute. +2. **Code** (*Run Once for All Items*), reads the cursor: + ```js + const data = $getWorkflowStaticData('global'); + return [{ json: { since: data.cursor || '15m' } }]; + ``` +3. **HTTP Request**: method `GET`, URL `https://logs.example.org/api/v1/logs`. + *Authentication*: Generic Credential Type > Header Auth, with name `Authorization` and value + `Bearer lst_…`. *Send Query Parameters*: `query` = `severity:in(emerg,alert,crit,err)`, + `since` = `{{ $json.since }}`, `limit` = `1000`. +4. **Code** (*Run Once for All Items*), saves the cursor and turns each log into an item: + ```js + const data = $getWorkflowStaticData('global'); + const page = $input.first().json; + data.cursor = page.next_cursor; + return page.rows.map((row) => ({ json: row })); + ``` + +Then connect what the logs should feed (Slack, mail, ticket, database…). n8n only keeps static +data for runs of the active workflow, not for manual test runs: while testing, each run starts +again from `15m`. When more than 1,000 logs arrive per minute, `has_more` is true and the +following runs catch up; raise `limit` or run the trigger more often if needed. + ## Settings The gear icon opens the settings, organized in tabs. Changes show at once but are only kept with the **Save** button, under the tab labels: it takes the accent color while changes are waiting. **Cancel** goes back to the saved settings, and closing the window with unsaved changes asks before dropping them. Localization and Appearance are remembered per browser; color tags -and log sources are saved on the server. Purging and resetting column widths act at once. +and log sources are saved on the server. Purging, API tokens and resetting column widths act at once. - **Localization** - *Language*: English or French. @@ -408,6 +477,9 @@ and log sources are saved on the server. Purging and resetting column widths act (already set in `docker-compose.yml`). The feature is off by default: set `ALLOW_PURGE=true` to allow it. Any admin can then purge: turn on [authentication](#authentication) if the UI is reachable by others. +- **API**: API tokens for the [reading API](#reading-api-n8n-scripts): create one with a name + (shown once, copy it then), see when and from where each was last used, revoke it. Reserved + to administrators. ## Authentication @@ -577,6 +649,8 @@ To update one of them: | `hub.go` | pushes new messages to browsers (SSE) | | `rdns.go` | cached reverse DNS lookups | | `export.go` | streamed CSV export | +| `logsapi.go` | reading API `/api/v1/logs`: cursor, groups of logs with the same time | +| `apitokens.go` | API tokens: storage (hashed), Bearer check, `/api/tokens` | | `docker.go` | Docker container logs (API, followers, positions, level detection) | | `syslogserver.go` | syslog listeners opened and closed from Settings > Log sources | | `hostlogs.go`, `journal.go` | host system logs: systemd journal reader (no `journalctl`) and `/var/log` follower | diff --git a/api.go b/api.go index 5bd2de5..b78fe52 100644 --- a/api.go +++ b/api.go @@ -28,11 +28,16 @@ type API struct { files *Files selfLog *SelfLog dbstats dbStatsCache + tokens *TokenStore } func (a *API) Routes(mux *http.ServeMux) { mux.HandleFunc("GET /healthz", a.health) mux.HandleFunc("GET /api/logs", a.logs) + mux.HandleFunc("GET /api/v1/logs", a.logsV1) + mux.HandleFunc("GET /api/tokens", a.listTokens) + mux.HandleFunc("POST /api/tokens", a.createToken) + mux.HandleFunc("DELETE /api/tokens/{id}", a.deleteToken) mux.HandleFunc("GET /api/histogram", a.histogram) mux.HandleFunc("GET /api/facets", a.facets) mux.HandleFunc("GET /api/stream", a.stream) diff --git a/apitokens.go b/apitokens.go new file mode 100644 index 0000000..51fa151 --- /dev/null +++ b/apitokens.go @@ -0,0 +1,254 @@ +package main + +import ( + "crypto/rand" + "crypto/sha256" + "crypto/subtle" + "encoding/base64" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "log" + "net/http" + "os" + "path/filepath" + "strings" + "sync" + "time" + "unicode/utf8" +) + +// API tokens: read-only access to /api/v1/ for scripts and automation tools +// (n8n…), sent as "Authorization: Bearer lst_…". They are created and revoked +// in Settings > API by an administrator, and work whatever the AUTH_MODE, +// including without authentication. Only a SHA-256 of each token is kept, in +// DATA_DIR/api_tokens.json: a token is shown once, when it is created. + +const ( + tokenPrefix = "lst_" + maxTokens = 100 + lastUsedPeriod = time.Minute // last use of a token is written to disk at most this often (or on a new IP) +) + +var errTokenNotFound = errors.New("token not found") + +type apiToken struct { + ID string `json:"id"` + Name string `json:"name"` + Hint string `json:"hint"` // start of the token, to recognize it + Hash string `json:"hash"` // hex SHA-256 of the whole token + Created time.Time `json:"created"` + LastUsed time.Time `json:"last_used,omitzero"` + LastIP string `json:"last_ip,omitempty"` +} + +// apiTokenInfo is what the UI sees of a token (never its hash). +type apiTokenInfo struct { + ID string `json:"id"` + Name string `json:"name"` + Hint string `json:"hint"` + Created time.Time `json:"created"` + LastUsed time.Time `json:"last_used,omitzero"` + LastIP string `json:"last_ip,omitempty"` +} + +type TokenStore struct { + path string + mu sync.Mutex + tokens []apiToken +} + +func LoadTokenStore(path string) (*TokenStore, error) { + s := &TokenStore{path: path} + b, err := os.ReadFile(path) + if errors.Is(err, os.ErrNotExist) { + return s, nil + } + if err != nil { + return nil, err + } + if err := json.Unmarshal(b, &s.tokens); err != nil { + return nil, fmt.Errorf("%s: %w", path, err) + } + return s, nil +} + +// save writes the file atomically, readable by its owner only; the caller holds the lock. +func (s *TokenStore) save() error { + b, err := json.MarshalIndent(s.tokens, "", " ") + if err != nil { + return err + } + if err := os.MkdirAll(filepath.Dir(s.path), 0o755); err != nil { + return err + } + tmp := s.path + ".tmp" + if err := os.WriteFile(tmp, b, 0o600); err != nil { + return err + } + return os.Rename(tmp, s.path) +} + +func hashToken(token string) string { + sum := sha256.Sum256([]byte(token)) + return hex.EncodeToString(sum[:]) +} + +func (s *TokenStore) List() []apiTokenInfo { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]apiTokenInfo, len(s.tokens)) + for i, t := range s.tokens { + out[i] = apiTokenInfo{ID: t.ID, Name: t.Name, Hint: t.Hint, Created: t.Created, LastUsed: t.LastUsed, LastIP: t.LastIP} + } + return out +} + +// Create makes a new token and returns it in clear, the only time it exists so. +func (s *TokenStore) Create(name string) (apiTokenInfo, string, error) { + name = strings.TrimSpace(name) + if name == "" || utf8.RuneCountInString(name) > 80 { + return apiTokenInfo{}, "", &codedError{code: "token_name", msg: "give the token a name of 1 to 80 characters"} + } + b := make([]byte, 32) + if _, err := rand.Read(b); err != nil { + return apiTokenInfo{}, "", err + } + secret := tokenPrefix + base64.RawURLEncoding.EncodeToString(b) + t := apiToken{ID: newID(), Name: name, Hint: secret[:len(tokenPrefix)+6], Hash: hashToken(secret), Created: time.Now().UTC()} + + s.mu.Lock() + defer s.mu.Unlock() + if len(s.tokens) >= maxTokens { + return apiTokenInfo{}, "", &codedError{code: "token_max", msg: fmt.Sprintf("at most %d API tokens", maxTokens)} + } + s.tokens = append(s.tokens, t) + if err := s.save(); err != nil { + s.tokens = s.tokens[:len(s.tokens)-1] + return apiTokenInfo{}, "", err + } + return apiTokenInfo{ID: t.ID, Name: t.Name, Hint: t.Hint, Created: t.Created}, secret, nil +} + +func (s *TokenStore) Delete(id string) error { + s.mu.Lock() + defer s.mu.Unlock() + for i, t := range s.tokens { + if t.ID == id { + s.tokens = append(s.tokens[:i:i], s.tokens[i+1:]...) + return s.save() + } + } + return errTokenNotFound +} + +// Check tells whether token is a valid API token, and notes its use. +func (s *TokenStore) Check(token, ip string) (name string, ok bool) { + if !strings.HasPrefix(token, tokenPrefix) { + return "", false + } + h := []byte(hashToken(token)) + s.mu.Lock() + defer s.mu.Unlock() + for i := range s.tokens { + t := &s.tokens[i] + if subtle.ConstantTimeCompare(h, []byte(t.Hash)) != 1 { + continue + } + now := time.Now().UTC() + stale := now.Sub(t.LastUsed) >= lastUsedPeriod || t.LastIP != ip + t.LastUsed, t.LastIP = now, ip + if stale { + if err := s.save(); err != nil { + log.Printf("api tokens: %v", err) + } + } + return t.Name, true + } + return "", false +} + +// bearerToken returns the token of an "Authorization: Bearer …" header. +func bearerToken(r *http.Request) (string, bool) { + scheme, token, ok := strings.Cut(r.Header.Get("Authorization"), " ") + if !ok || !strings.EqualFold(scheme, "Bearer") { + return "", false + } + return strings.TrimSpace(token), true +} + +// tokenAuth serves the requests carrying an API token itself, read-only and +// limited to /api/v1/, before the session login (auth); api is the handler +// behind the login. +func tokenAuth(tokens *TokenStore, auth, api http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + token, ok := bearerToken(r) + if !ok { + auth.ServeHTTP(w, r) + return + } + ip := clientIP(r) + if _, valid := tokens.Check(token, ip); !valid { + log.Printf("api: invalid or revoked token from %s", ip) + time.Sleep(loginFailDelay) + w.Header().Set("WWW-Authenticate", `Bearer realm="logstream"`) + writeErr(w, http.StatusUnauthorized, &codedError{code: "token", msg: "invalid or revoked API token"}) + return + } + if !strings.HasPrefix(r.URL.Path, "/api/v1/") || !isSafeMethod(r.Method) { + writeErr(w, http.StatusForbidden, &codedError{code: "token_scope", msg: "API tokens only give read access to /api/v1/"}) + return + } + api.ServeHTTP(w, asViewer(r)) + }) +} + +// GET /api/tokens: the API tokens (Settings > API), for administrators. +func (a *API) listTokens(w http.ResponseWriter, r *http.Request) { + if isViewer(r) { + writeErr(w, http.StatusForbidden, &codedError{code: "read_only", msg: "read-only account: changes are reserved to administrators"}) + return + } + writeJSON(w, http.StatusOK, a.tokens.List()) +} + +// POST /api/tokens {"name": "n8n"}: returns the new token, with "token" in clear. +func (a *API) createToken(w http.ResponseWriter, r *http.Request) { + var body struct { + Name string `json:"name"` + } + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4096)).Decode(&body); err != nil { + writeErr(w, http.StatusBadRequest, err) + return + } + info, secret, err := a.tokens.Create(body.Name) + var ce *codedError + switch { + case errors.As(err, &ce): + writeErr(w, http.StatusBadRequest, err) + return + case err != nil: + writeErr(w, http.StatusInternalServerError, err) + return + } + log.Printf("api token %q created from %s", info.Name, clientIP(r)) + writeJSON(w, http.StatusCreated, struct { + apiTokenInfo + Token string `json:"token"` + }{info, secret}) +} + +// DELETE /api/tokens/{id}: revokes a token at once. +func (a *API) deleteToken(w http.ResponseWriter, r *http.Request) { + err := a.tokens.Delete(r.PathValue("id")) + switch { + case errors.Is(err, errTokenNotFound): + writeErr(w, http.StatusNotFound, err) + case err != nil: + writeErr(w, http.StatusInternalServerError, err) + default: + log.Printf("api token %s revoked from %s", r.PathValue("id"), clientIP(r)) + w.WriteHeader(http.StatusNoContent) + } +} diff --git a/apitokens_test.go b/apitokens_test.go new file mode 100644 index 0000000..7e332c1 --- /dev/null +++ b/apitokens_test.go @@ -0,0 +1,176 @@ +package main + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestTokenStore(t *testing.T) { + path := filepath.Join(t.TempDir(), "api_tokens.json") + s, err := LoadTokenStore(path) + if err != nil { + t.Fatal(err) + } + if _, _, err := s.Create(" "); err == nil { + t.Fatal("a token without a name should be refused") + } + info, secret, err := s.Create("n8n") + if err != nil { + t.Fatal(err) + } + if !strings.HasPrefix(secret, tokenPrefix) || len(secret) < 40 || !strings.HasPrefix(secret, info.Hint) { + t.Fatalf("token %q, hint %q", secret, info.Hint) + } + b, _ := os.ReadFile(path) + if strings.Contains(string(b), secret) || !strings.Contains(string(b), hashToken(secret)) { + t.Fatal("the file must hold the hash of the token, not the token") + } + if st, _ := os.Stat(path); st.Mode().Perm() != 0o600 { + t.Fatalf("file mode %v, want 0600", st.Mode().Perm()) + } + if _, ok := s.Check(secret+"x", "10.0.0.1"); ok { + t.Fatal("a wrong token was accepted") + } + if name, ok := s.Check(secret, "10.0.0.1"); !ok || name != "n8n" { + t.Fatal("the token was refused") + } + + // Reloaded from disk (restart): still valid, last use kept. + s2, err := LoadTokenStore(path) + if err != nil { + t.Fatal(err) + } + if l := s2.List(); len(l) != 1 || l[0].LastIP != "10.0.0.1" || l[0].LastUsed.IsZero() { + t.Fatalf("after reload: %+v", l) + } + if _, ok := s2.Check(secret, "10.0.0.2"); !ok { + t.Fatal("token refused after reload") + } + if err := s2.Delete(info.ID); err != nil { + t.Fatal(err) + } + if _, ok := s2.Check(secret, "10.0.0.2"); ok { + t.Fatal("a revoked token was accepted") + } + if err := s2.Delete(info.ID); err != errTokenNotFound { + t.Fatalf("second delete: %v", err) + } +} + +func TestTokenAuth(t *testing.T) { + loginFailDelay = 0 + tokens, _ := LoadTokenStore(filepath.Join(t.TempDir(), "api_tokens.json")) + _, secret, _ := tokens.Create("n8n") + + var reached string + mux := http.NewServeMux() + mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + reached = r.URL.Path + if isViewer(r) { + reached += " viewer" + } + }) + app := readOnly(mux) + login := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { writeAuthRequired(w) }) + h := tokenAuth(tokens, login, app) + + call := func(method, path, auth string) int { + reached = "" + req := httptest.NewRequest(method, path, nil) + if auth != "" { + req.Header.Set("Authorization", auth) + } + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + return rec.Code + } + if code := call("GET", "/api/v1/logs", "Bearer "+secret); code != http.StatusOK || reached != "/api/v1/logs viewer" { + t.Fatalf("valid token: %d, reached %q", code, reached) + } + if code := call("GET", "/api/v1/logs", "bearer "+secret); code != http.StatusOK { + t.Fatalf("lower-case scheme: %d", code) + } + for _, tc := range []struct { + method, path, auth string + want int + }{ + {"GET", "/api/v1/logs", "Bearer lst_wrong", http.StatusUnauthorized}, + {"GET", "/api/v1/logs", "Bearer ", http.StatusUnauthorized}, + {"GET", "/api/tags", "Bearer " + secret, http.StatusForbidden}, + {"GET", "/api/tokens", "Bearer " + secret, http.StatusForbidden}, + {"POST", "/api/v1/logs", "Bearer " + secret, http.StatusForbidden}, + {"GET", "/api/v1/logs", "", http.StatusUnauthorized}, // no token: the login decides + {"GET", "/api/v1/logs", "Basic dTpw", http.StatusUnauthorized}, // Basic goes to the login + } { + if code := call(tc.method, tc.path, tc.auth); code != tc.want || reached != "" { + t.Errorf("%s %s %q: %d (reached %q), want %d", tc.method, tc.path, tc.auth, code, reached, tc.want) + } + } +} + +// With local authentication, a token opens /api/v1/ without a session. +func TestTokenWithLocalAuth(t *testing.T) { + loginFailDelay = 0 + dir := t.TempDir() + tokens, _ := LoadTokenStore(filepath.Join(dir, "api_tokens.json")) + _, secret, _ := tokens.Create("script") + app := readOnly(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write([]byte("ok")) })) + auth, err := newAuth(authConfig{mode: "local", user: "admin", pass: "pw", dataDir: dir, sessionTTL: time.Hour}, app) + if err != nil { + t.Fatal(err) + } + h := tokenAuth(tokens, auth, app) + req := httptest.NewRequest("GET", "/api/v1/logs", nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusUnauthorized { + t.Fatalf("without token: %d", rec.Code) + } + req.Header.Set("Authorization", "Bearer "+secret) + rec = httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusOK || rec.Body.String() != "ok" { + t.Fatalf("with token: %d %s", rec.Code, rec.Body) + } +} + +func TestTokenHandlers(t *testing.T) { + tokens, _ := LoadTokenStore(filepath.Join(t.TempDir(), "api_tokens.json")) + a := &API{tokens: tokens} + mux := http.NewServeMux() + a.Routes(mux) + do := func(r *http.Request) *httptest.ResponseRecorder { + rec := httptest.NewRecorder() + readOnly(mux).ServeHTTP(rec, r) + return rec + } + rec := do(httptest.NewRequest("POST", "/api/tokens", strings.NewReader(`{"name":"n8n"}`))) + var created struct { + ID, Token, Hint string + } + if err := json.Unmarshal(rec.Body.Bytes(), &created); rec.Code != http.StatusCreated || err != nil || created.Token == "" { + t.Fatalf("create: %d %s", rec.Code, rec.Body) + } + rec = do(httptest.NewRequest("GET", "/api/tokens", nil)) + if rec.Code != http.StatusOK || strings.Contains(rec.Body.String(), created.Token) || strings.Contains(rec.Body.String(), "hash") { + t.Fatalf("list must not show the token or its hash: %s", rec.Body) + } + if rec := do(asViewer(httptest.NewRequest("GET", "/api/tokens", nil))); rec.Code != http.StatusForbidden { + t.Fatalf("viewer list: %d", rec.Code) + } + if rec := do(asViewer(httptest.NewRequest("DELETE", "/api/tokens/"+created.ID, nil))); rec.Code != http.StatusForbidden { + t.Fatalf("viewer delete: %d", rec.Code) + } + if rec := do(httptest.NewRequest("DELETE", "/api/tokens/"+created.ID, nil)); rec.Code != http.StatusNoContent { + t.Fatalf("delete: %d", rec.Code) + } + if rec := do(httptest.NewRequest("DELETE", "/api/tokens/"+created.ID, nil)); rec.Code != http.StatusNotFound { + t.Fatalf("delete again: %d", rec.Code) + } +} diff --git a/logsapi.go b/logsapi.go new file mode 100644 index 0000000..4e7342c --- /dev/null +++ b/logsapi.go @@ -0,0 +1,281 @@ +package main + +import ( + "context" + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "net/http" + "sort" + "strconv" + "strings" + "time" +) + +// GET /api/v1/logs: incremental reading of the stored logs for automation +// tools (n8n, scripts). Each answer gives the rows that follow a cursor, +// oldest first, and the cursor to send next time; following the cursors +// returns every row once, without gaps or duplicates. +// +// The cursor is a position on _time, which is the reception time given by +// LogStream (see Entry.Record), so new rows always come after it. Rows only +// count once they are older than the lag (30 s by default): a message can +// wait a little in the queue or for its DNS name before VictoriaLogs makes it +// visible. Several rows can share the same _time to the nanosecond; inside +// such a group the rows are ordered by a hash of their content, and the +// cursor remembers how many of them were already returned. + +const ( + logsAPIDefaultLimit = 1000 + logsAPIMaxLimit = 10000 + logsAPIDefaultLag = 30 * time.Second + logsAPIMaxLag = time.Hour + logsAPIDefaultSince = time.Hour + logsAPIMaxGroup = 100000 // rows with the very same _time +) + +// logsCursor: every row with _time < T has been returned, plus the first K +// rows (in hash order) of those with _time == T. Q is a hash of the query: +// K only means something for the same query. +type logsCursor struct { + T time.Time `json:"t"` + K int `json:"k,omitempty"` + Q string `json:"q,omitempty"` +} + +func (c logsCursor) encode() string { + b, _ := json.Marshal(c) + return base64.RawURLEncoding.EncodeToString(b) +} + +func decodeCursor(s string) (logsCursor, bool) { + var c logsCursor + b, err := base64.RawURLEncoding.DecodeString(s) + if err != nil || json.Unmarshal(b, &c) != nil || c.T.IsZero() || c.K < 0 { + return logsCursor{}, false + } + return c, true +} + +// parseSince reads the since parameter: a cursor from a previous answer, an +// RFC 3339 time, or a duration back from now (90s, 15m, 6h, 7d). Empty: the +// last hour. +func parseSince(s string, now time.Time) (logsCursor, error) { + s = strings.TrimSpace(s) + if s == "" { + return logsCursor{T: now.Add(-logsAPIDefaultSince)}, nil + } + if t, err := time.Parse(time.RFC3339Nano, s); err == nil { + return logsCursor{T: t}, nil + } + if d, ok := parseAge(s); ok { + return logsCursor{T: now.Add(-d)}, nil + } + if c, ok := decodeCursor(s); ok { + return c, nil + } + return logsCursor{}, &codedError{code: "bad_since", msg: "since: expected a next_cursor, an RFC 3339 time or a duration such as 15m, 6h or 7d"} +} + +// parseAge reads a Go duration, plus "Nd" for days. +func parseAge(s string) (time.Duration, bool) { + if n, ok := strings.CutSuffix(s, "d"); ok { + days, err := strconv.Atoi(n) + if err != nil || days < 0 { + return 0, false + } + return time.Duration(days) * 24 * time.Hour, true + } + d, err := time.ParseDuration(s) + return d, err == nil && d >= 0 +} + +type logsPage struct { + Rows []map[string]any `json:"rows"` + NextCursor string `json:"next_cursor"` + HasMore bool `json:"has_more"` // call again right away: more rows are ready + Waiting bool `json:"waiting,omitempty"` // the disk buffer is still being sent to VictoriaLogs + Until string `json:"until"` // rows are returned up to this time (now - lag) +} + +// GET /api/v1/logs?query=&since=&limit=&lag= +func (a *API) logsV1(w http.ResponseWriter, r *http.Request) { + q := r.URL.Query() + filter := strings.TrimSpace(q.Get("query")) + if f, pipes := splitPipes(filter); strings.TrimSpace(pipes) != "" { + writeErr(w, http.StatusBadRequest, &codedError{code: "api_pipes", msg: "query: LogsQL pipes (| ...) are not supported, keep only the filter part"}) + return + } else { + filter = strings.TrimSpace(f) + } + if filter == "" { + filter = "*" + } + limit := logsAPIDefaultLimit + if v := q.Get("limit"); v != "" { + n, err := strconv.Atoi(v) + if err != nil || n <= 0 || n > logsAPIMaxLimit { + writeErr(w, http.StatusBadRequest, &codedError{code: "bad_limit", msg: fmt.Sprintf("limit: expected 1 to %d", logsAPIMaxLimit)}) + return + } + limit = n + } + lag := logsAPIDefaultLag + if v := q.Get("lag"); v != "" { + d, ok := parseAge(v) + if !ok || d > logsAPIMaxLag { + writeErr(w, http.StatusBadRequest, &codedError{code: "bad_lag", msg: "lag: expected a duration from 0s to 1h"}) + return + } + lag = d + } + now := time.Now() + cur, err := parseSince(q.Get("since"), now) + if err != nil { + writeErr(w, http.StatusBadRequest, err) + return + } + qh := queryHash(filter) + if cur.Q != qh { + cur.K = 0 // the query changed: the rank inside the group no longer applies + } + cur.Q = qh + + // Logs waiting in the disk buffer were received earlier: returning newer + // rows now would move the cursor past them. + if a.store.PendingSpool() > 0 { + writeJSON(w, http.StatusOK, logsPage{Rows: []map[string]any{}, NextCursor: cur.encode(), Waiting: true, Until: logsqlTime(cur.T)}) + return + } + until := now.Add(-lag) + page, err := readLogsPage(r.Context(), a.store.Query, filter, cur, until, limit) + if err != nil { + writeErr(w, queryStatus(err), err) + return + } + a.annotateHosts(page.Rows) + writeJSON(w, http.StatusOK, page) +} + +func queryHash(filter string) string { + sum := sha256.Sum256([]byte(filter)) + return hex.EncodeToString(sum[:4]) +} + +type queryFunc func(ctx context.Context, query string) ([]map[string]any, error) + +// readLogsPage returns at most limit rows after cur, up to until included. +func readLogsPage(ctx context.Context, run queryFunc, filter string, cur logsCursor, until time.Time, limit int) (logsPage, error) { + page := logsPage{Rows: []map[string]any{}, Until: logsqlTime(until)} + next := cur + start := cur.T + if cur.K > 0 { + // Rest of the group the previous page stopped in. + group, err := readGroup(ctx, run, filter, cur.T) + if err != nil { + return page, err + } + rest := group[min(cur.K, len(group)):] + if len(rest) >= limit { + page.Rows = rest[:limit] + next.K = cur.K + limit + page.HasMore = true + page.NextCursor = next.encode() + return page, nil + } + page.Rows = append(page.Rows, rest...) + start = cur.T.Add(time.Nanosecond) + } + + left := limit - len(page.Rows) + var rows []map[string]any + if !start.After(until) { + q := fmt.Sprintf("_time:[%s, %s] (%s) | sort by (_time) | limit %d", logsqlTime(start), logsqlTime(until), filter, left+1) + var err error + if rows, err = run(ctx, q); err != nil { + return page, err + } + } + if len(rows) <= left { + // Everything up to until has been read. + page.Rows = append(page.Rows, rows...) + next = logsCursor{T: until.Add(time.Nanosecond), Q: cur.Q} + if next.T.Before(start) { + next.T = start + } + page.NextCursor = next.encode() + return page, nil + } + + // One row too many: the group of the last time may be cut. Whole groups + // before it are returned; that last group starts the next page. + page.HasMore = true + last, err := rowTime(rows[left]) + if err != nil { + return page, err + } + n := 0 + for n < left { + t, err := rowTime(rows[n]) + if err != nil { + return page, err + } + if !t.Before(last) { + break + } + n++ + } + if n > 0 || len(page.Rows) > 0 { + page.Rows = append(page.Rows, rows[:n]...) + page.NextCursor = logsCursor{T: last, Q: cur.Q}.encode() + return page, nil + } + // The whole page is a single group: read it all to cut it in a stable order. + group, err := readGroup(ctx, run, filter, last) + if err != nil { + return page, err + } + page.Rows = group[:min(left, len(group))] + page.NextCursor = logsCursor{T: last, K: len(page.Rows), Q: cur.Q}.encode() + return page, nil +} + +// readGroup returns the rows whose _time is exactly t, in a stable order. +func readGroup(ctx context.Context, run queryFunc, filter string, t time.Time) ([]map[string]any, error) { + ts := logsqlTime(t) + rows, err := run(ctx, fmt.Sprintf("_time:[%s, %s] (%s) | limit %d", ts, ts, filter, logsAPIMaxGroup+1)) + if err != nil { + return nil, err + } + if len(rows) > logsAPIMaxGroup { + return nil, fmt.Errorf("more than %d logs share the time %s", logsAPIMaxGroup, ts) + } + keys := make([]string, len(rows)) + for i, row := range rows { + b, _ := json.Marshal(row) // map keys are sorted: same row, same bytes + sum := sha256.Sum256(b) + keys[i] = string(sum[:]) + } + idx := make([]int, len(rows)) + for i := range idx { + idx[i] = i + } + sort.SliceStable(idx, func(i, j int) bool { return keys[idx[i]] < keys[idx[j]] }) + out := make([]map[string]any, len(rows)) + for i, k := range idx { + out[i] = rows[k] + } + return out, nil +} + +func rowTime(row map[string]any) (time.Time, error) { + s, _ := row["_time"].(string) + t, err := time.Parse(time.RFC3339Nano, s) + if err != nil { + return time.Time{}, errors.New("VictoriaLogs returned a row without a valid _time") + } + return t, nil +} diff --git a/logsapi_test.go b/logsapi_test.go new file mode 100644 index 0000000..a7ed844 --- /dev/null +++ b/logsapi_test.go @@ -0,0 +1,276 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "io" + "math/rand" + "net/http" + "net/http/httptest" + "regexp" + "sort" + "strconv" + "strings" + "sync" + "testing" + "time" +) + +// cursorVL plays VictoriaLogs for the two queries of /api/v1/logs: a time +// range sorted by _time, and an exact time. Like VictoriaLogs, it gives no +// order among rows of the same _time: they are shuffled on every query. +type cursorVL struct { + mu sync.Mutex + rows []map[string]any + rnd *rand.Rand + queries []string +} + +var cursorQuery = regexp.MustCompile(`^_time:\[([^,]+), ([^\]]+)] \((.*?)\)( \| sort by \(_time\))? \| limit (\d+)$`) + +func (v *cursorVL) add(t time.Time, msg string) { + v.mu.Lock() + defer v.mu.Unlock() + v.rows = append(v.rows, map[string]any{"_time": t.UTC().Format(time.RFC3339Nano), "_msg": msg, "host": "h1"}) +} + +func (v *cursorVL) RoundTrip(r *http.Request) (*http.Response, error) { + _ = r.ParseForm() + q := r.PostForm.Get("query") + m := cursorQuery.FindStringSubmatch(q) + if m == nil { + return &http.Response{StatusCode: http.StatusBadRequest, Body: io.NopCloser(strings.NewReader("unexpected query " + q)), Header: http.Header{}}, nil + } + from, _ := time.Parse(time.RFC3339Nano, m[1]) + to, _ := time.Parse(time.RFC3339Nano, m[2]) + limit, _ := strconv.Atoi(m[5]) + + v.mu.Lock() + defer v.mu.Unlock() + v.queries = append(v.queries, q) + var out []map[string]any + for _, row := range v.rows { + t, _ := time.Parse(time.RFC3339Nano, row["_time"].(string)) + if !t.Before(from) && !t.After(to) { + out = append(out, row) + } + } + v.rnd.Shuffle(len(out), func(i, j int) { out[i], out[j] = out[j], out[i] }) + if m[4] != "" { + sort.SliceStable(out, func(i, j int) bool { + a, _ := time.Parse(time.RFC3339Nano, out[i]["_time"].(string)) + b, _ := time.Parse(time.RFC3339Nano, out[j]["_time"].(string)) + return a.Before(b) + }) + } + if len(out) > limit { + out = out[:limit] + } + var b strings.Builder + for _, row := range out { + line, _ := json.Marshal(row) + b.Write(line) + b.WriteByte('\n') + } + return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(b.String())), Header: http.Header{}}, nil +} + +func newCursorVL(seed int64) (*cursorVL, *Store) { + vl := &cursorVL{rnd: rand.New(rand.NewSource(seed))} + store := NewStore("http://vl", 10, 10, time.Second, nil) + store.client.Transport = vl + store.streamClient.Transport = vl + return vl, store +} + +// fill adds rows by groups of the same _time, some groups larger than a +// page, some with identical rows, and returns the multiset of the rows. +func fill(vl *cursorVL, start time.Time, groups []int, tag string) map[string]int { + want := map[string]int{} + t := start + for g, size := range groups { + for i := 0; i < size; i++ { + msg := fmt.Sprintf("%s g%d r%d", tag, g, i%4) // r0..r3 repeat: identical rows + vl.add(t, msg) + want[rowKey(t, msg)]++ + } + t = t.Add(time.Duration(1+g%3) * time.Microsecond) + } + return want +} + +func rowKey(t time.Time, msg string) string { return t.UTC().Format(time.RFC3339Nano) + " " + msg } + +func TestLogsCursorNoLossNoDuplicate(t *testing.T) { + base := time.Date(2026, 11, 1, 10, 0, 0, 0, time.UTC) + groups := []int{1, 3, 1, 1, 7, 2, 1, 12, 1, 4, 4, 1, 25, 1, 2} + for _, limit := range []int{1, 2, 3, 5, 8, 1000} { + for seed := int64(1); seed <= 5; seed++ { + vl, store := newCursorVL(seed) + want := fill(vl, base, groups, "a") + got := map[string]int{} + cur := logsCursor{T: base.Add(-time.Minute)} + until := base.Add(time.Second) + var last time.Time + for calls := 0; ; calls++ { + if calls > 500 { + t.Fatalf("limit %d: no end", limit) + } + page, err := readLogsPage(context.Background(), store.Query, "*", cur, until, limit) + if err != nil { + t.Fatal(err) + } + if len(page.Rows) > limit { + t.Fatalf("limit %d: page of %d rows", limit, len(page.Rows)) + } + for _, row := range page.Rows { + rt, _ := rowTime(row) + if rt.Before(last) { + t.Fatalf("limit %d: rows out of order (%s after %s)", limit, rt, last) + } + last = rt + got[rowKey(rt, row["_msg"].(string))]++ + } + var ok bool + if cur, ok = decodeCursor(page.NextCursor); !ok { + t.Fatalf("bad cursor %q", page.NextCursor) + } + if !page.HasMore { + break + } + if len(page.Rows) == 0 { + t.Fatalf("limit %d: has_more with an empty page", limit) + } + } + if fmt.Sprint(got) != fmt.Sprint(want) { + t.Fatalf("limit %d seed %d:\n got %v\nwant %v", limit, seed, got, want) + } + } + } +} + +// New logs arrive between the calls, as with a scheduled n8n workflow. +func TestLogsCursorFollowsNewLogs(t *testing.T) { + vl, store := newCursorVL(7) + base := time.Date(2026, 11, 1, 10, 0, 0, 0, time.UTC) + want := map[string]int{} + got := map[string]int{} + cur := logsCursor{T: base} + for round := 0; round < 6; round++ { + start := base.Add(time.Duration(round) * time.Second) + for k, n := range fill(vl, start, []int{2, 5, 1, 3}, fmt.Sprintf("round%d", round)) { + want[k] += n + } + until := start.Add(500 * time.Millisecond) // after this round's rows + for { + page, err := readLogsPage(context.Background(), store.Query, "*", cur, until, 4) + if err != nil { + t.Fatal(err) + } + for _, row := range page.Rows { + rt, _ := rowTime(row) + got[rowKey(rt, row["_msg"].(string))]++ + } + cur, _ = decodeCursor(page.NextCursor) + if !page.HasMore { + break + } + } + // Polling again with nothing new returns nothing. + page, err := readLogsPage(context.Background(), store.Query, "*", cur, until, 4) + if err != nil || len(page.Rows) != 0 || page.HasMore { + t.Fatalf("round %d: second poll returned %d rows (has_more %v, err %v)", round, len(page.Rows), page.HasMore, err) + } + } + if fmt.Sprint(got) != fmt.Sprint(want) { + t.Fatalf("got %v\nwant %v", got, want) + } +} + +func TestParseSince(t *testing.T) { + now := time.Date(2026, 11, 1, 12, 0, 0, 0, time.UTC) + c := logsCursor{T: now.Add(-time.Minute), K: 3, Q: "abcd"} + for _, tc := range []struct { + in string + want logsCursor + }{ + {"", logsCursor{T: now.Add(-time.Hour)}}, + {"15m", logsCursor{T: now.Add(-15 * time.Minute)}}, + {"7d", logsCursor{T: now.Add(-7 * 24 * time.Hour)}}, + {"2026-11-01T08:30:00.5Z", logsCursor{T: time.Date(2026, 11, 1, 8, 30, 0, 5e8, time.UTC)}}, + {c.encode(), c}, + } { + got, err := parseSince(tc.in, now) + if err != nil || !got.T.Equal(tc.want.T) || got.K != tc.want.K || got.Q != tc.want.Q { + t.Errorf("parseSince(%q) = %+v, %v; want %+v", tc.in, got, err, tc.want) + } + } + for _, bad := range []string{"yesterday", "-5m", "e30"} { + if _, err := parseSince(bad, now); err == nil { + t.Errorf("parseSince(%q): expected an error", bad) + } + } +} + +func TestLogsV1Handler(t *testing.T) { + vl, store := newCursorVL(3) + a := &API{store: store, rdns: NewReverseDNS(false, "")} + old := time.Now().Add(-10 * time.Minute) + fill(vl, old, []int{2, 1}, "x") + vl.add(time.Now().Add(time.Hour), "in the future: after until") + + get := func(url string) (int, logsPage, map[string]string) { + rec := httptest.NewRecorder() + a.logsV1(rec, httptest.NewRequest("GET", url, nil)) + var page logsPage + var e map[string]string + _ = json.Unmarshal(rec.Body.Bytes(), &page) + _ = json.Unmarshal(rec.Body.Bytes(), &e) + return rec.Code, page, e + } + code, page, _ := get("/api/v1/logs?query=host:h1&since=1h") + if code != http.StatusOK || len(page.Rows) != 3 || page.HasMore || page.NextCursor == "" { + t.Fatalf("status %d, %d rows, has_more %v", code, len(page.Rows), page.HasMore) + } + if q := vl.queries[len(vl.queries)-1]; !strings.Contains(q, "(host:h1)") { + t.Fatalf("query filter not applied: %s", q) + } + code, page, _ = get("/api/v1/logs?query=host:h1&since=" + page.NextCursor) + if code != http.StatusOK || len(page.Rows) != 0 { + t.Fatalf("second call: status %d, %d rows", code, len(page.Rows)) + } + for url, wantCode := range map[string]string{ + "/api/v1/logs?query=" + "*%20|%20stats%20count()": "api_pipes", + "/api/v1/logs?limit=0": "bad_limit", + "/api/v1/logs?limit=20000": "bad_limit", + "/api/v1/logs?lag=2h": "bad_lag", + "/api/v1/logs?since=nope": "bad_since", + } { + if code, _, e := get(url); code != http.StatusBadRequest || e["code"] != wantCode { + t.Errorf("%s: status %d, %v; want 400 %s", url, code, e, wantCode) + } + } +} + +func TestLogsV1WaitsForSpool(t *testing.T) { + spool, err := OpenSpool(t.TempDir(), 1<<20) + if err != nil { + t.Fatal(err) + } + if err := spool.Write([]byte(`{"_msg":"x"}`+"\n"), 1); err != nil { + t.Fatal(err) + } + vl, _ := newCursorVL(1) + store := NewStore("http://vl", 10, 10, time.Second, spool) + store.streamClient.Transport = vl + a := &API{store: store} + rec := httptest.NewRecorder() + a.logsV1(rec, httptest.NewRequest("GET", "/api/v1/logs?since=2026-11-01T10:00:00Z", nil)) + var page logsPage + _ = json.Unmarshal(rec.Body.Bytes(), &page) + c, _ := decodeCursor(page.NextCursor) + if rec.Code != http.StatusOK || !page.Waiting || len(page.Rows) != 0 || !c.T.Equal(time.Date(2026, 11, 1, 10, 0, 0, 0, time.UTC)) || len(vl.queries) != 0 { + t.Fatalf("status %d, waiting %v, %d rows, cursor %v, %d queries", rec.Code, page.Waiting, len(page.Rows), c.T, len(vl.queries)) + } +} diff --git a/main.go b/main.go index 7d76e91..22c8bc6 100644 --- a/main.go +++ b/main.go @@ -142,6 +142,10 @@ func main() { if err != nil { log.Fatalf("tags: %v", err) } + tokens, err := LoadTokenStore(filepath.Join(cfg.dataDir, "api_tokens.json")) + if err != nil { + log.Fatalf("api tokens: %v", err) + } var spool *Spool if cfg.spoolMax > 0 { @@ -200,7 +204,7 @@ func main() { } mux := http.NewServeMux() presets := getenv("PRESETS_FILE", filepath.Join(cfg.dataDir, "presets.json")) - api := &API{store: store, hub: hub, tags: tags, presets: presets, rdns: rdns, allowPurge: cfg.allowPurge, exportMax: cfg.exportMax, syslog: syslogSrv} + api := &API{store: store, hub: hub, tags: tags, presets: presets, rdns: rdns, allowPurge: cfg.allowPurge, exportMax: cfg.exportMax, syslog: syslogSrv, tokens: tokens} if cfg.dockerLogs { dm, err := NewDockerManager(cfg.dockerHost, cfg.dataDir, cfg.backfill, sink) if err != nil { @@ -228,7 +232,8 @@ func main() { api.Routes(mux) mux.Handle("GET /", http.FileServer(http.FS(static))) - handler, err := newAuth(cfg.auth, readOnly(mux)) + app := readOnly(mux) + handler, err := newAuth(cfg.auth, app) if err != nil { log.Fatalf("auth: %v", err) } @@ -241,7 +246,8 @@ func main() { if !authOn { log.Printf("warning: no authentication (AUTH_USER is empty): anyone who can reach %s can read the logs and change the settings", cfg.httpAddr) } - handler = secure(handler, contentSecurityPolicy(static), authOn) + // API tokens (Settings > API) are checked before the login, in every auth mode. + handler = secure(tokenAuth(tokens, handler, app), contentSecurityPolicy(static), authOn) srv := &http.Server{ Addr: cfg.httpAddr, Handler: handler, diff --git a/store.go b/store.go index e1a2c02..e036fb9 100644 --- a/store.go +++ b/store.go @@ -423,6 +423,15 @@ func queryStatus(err error) int { return http.StatusBadGateway } +// PendingSpool is the number of logs waiting in the disk spool. +func (s *Store) PendingSpool() int64 { + if s.spool == nil { + return 0 + } + lines, _ := s.spool.Pending() + return lines +} + func (s *Store) Stats() map[string]any { st := map[string]any{ "received": s.received.Load(), diff --git a/web/app.js b/web/app.js index abce485..763da75 100644 --- a/web/app.js +++ b/web/app.js @@ -174,6 +174,25 @@ const I18N = { exportNote: (n) => `Up to ${n} logs matching the current filters, newest first. Dates use the time zone chosen in Settings.`, exporting: 'Exporting…', exportDone: (f) => `Downloaded: ${f}`, exportError: 'Export failed: ', err_export_pipes: 'Export does not support LogsQL pipes (| …): keep only the filter part', + tabApi: 'API', + tokensTitle: 'API tokens', + tokensHelp: 'Tokens give scripts and automation tools (n8n…) read-only access to /api/v1/logs, whatever the authentication mode. Only a hash is kept: a token is shown once, when it is created. Revoking it cuts off access at once.', + tokenCount: (n) => `${n} token${n > 1 ? 's' : ''}`, + tokenNone: 'No API token yet.', + tokenNamePh: 'Token name (n8n, backup script…)', + tokenCreate: 'Create a token', + tokenOnce: 'Copy this token now: it will not be shown again.', + tokenCopy: 'Copy', + tokenRevoke: 'Revoke', + tokenRevokeAsk: (n) => `Revoke the token "${n}"? Tools using it lose access at once.`, + tokenRevoked: 'Token revoked', + tokenCreated: (d) => `created ${d}`, + tokenUsed: (p) => `last used ${p.d}${p.ip ? ' from ' + p.ip : ''}`, + tokenNever: 'never used', + tokensUse: 'Usage', + tokensUseHelp: 'Send the token in an Authorization header. Each answer has next_cursor: pass it as since on the next call to get the following logs, without gaps or duplicates (see the README for an n8n workflow).', + err_token_name: 'Give the token a name of 1 to 80 characters', + err_token_max: 'Too many API tokens', tabLocale: 'Localization', tabFilters: 'Patterns', tabInterface: 'Appearance', tabData: 'Database', themeTitle: 'Theme', themeSystem: 'System', themeLight: 'Light', themeDark: 'Dark', themeHelp: 'System follows the light/dark preference of your computer or phone.', @@ -406,6 +425,25 @@ const I18N = { exportNote: (n) => `Jusqu'à ${n} logs correspondant aux filtres, du plus récent au plus ancien. Dates dans le fuseau choisi dans Paramètres.`, exporting: 'Export…', exportDone: (f) => `Téléchargé : ${f}`, exportError: 'Échec de l\'export : ', err_export_pipes: 'L\'export ne gère pas les pipes LogsQL (| …) : gardez seulement la partie filtre', + tabApi: 'API', + tokensTitle: "Jetons d'API", + tokensHelp: "Les jetons donnent aux scripts et outils d'automatisation (n8n…) un accès en lecture seule à /api/v1/logs, quel que soit le mode d'authentification. Seule une empreinte est conservée : un jeton n'est affiché qu'une fois, à sa création. Le révoquer coupe l'accès immédiatement.", + tokenCount: (n) => `${n} jeton${n > 1 ? 's' : ''}`, + tokenNone: "Aucun jeton d'API pour l'instant.", + tokenNamePh: 'Nom du jeton (n8n, script de sauvegarde…)', + tokenCreate: 'Créer un jeton', + tokenOnce: "Copiez ce jeton maintenant : il ne sera plus affiché.", + tokenCopy: 'Copier', + tokenRevoke: 'Révoquer', + tokenRevokeAsk: (n) => `Révoquer le jeton « ${n} » ? Les outils qui l'utilisent perdent l'accès immédiatement.`, + tokenRevoked: 'Jeton révoqué', + tokenCreated: (d) => `créé le ${d}`, + tokenUsed: (p) => `dernière utilisation ${p.d}${p.ip ? ' depuis ' + p.ip : ''}`, + tokenNever: 'jamais utilisé', + tokensUse: 'Utilisation', + tokensUseHelp: "Envoyez le jeton dans un en-tête Authorization. Chaque réponse contient next_cursor : passez-le dans since à l'appel suivant pour obtenir les logs qui suivent, sans trou ni doublon (voir le README pour un workflow n8n).", + err_token_name: 'Donnez au jeton un nom de 1 à 80 caractères', + err_token_max: "Trop de jetons d'API", tabLocale: 'Localisation et langue', tabFilters: 'Motifs de recherche', tabInterface: 'Affichage', tabData: 'Base de données', themeTitle: 'Thème', themeSystem: 'Système', themeLight: 'Clair', themeDark: 'Sombre', themeHelp: 'Système suit la préférence clair/sombre de votre ordinateur ou de votre téléphone.', @@ -2664,6 +2702,74 @@ $('#purgeBtn').addEventListener('click', async () => { } }); +/* ================= API tokens ================= */ + +const tokens = { list: null, error: null }; + +function renderTokens() { + const el = $('#tokenList'); + const day = (x) => fmtFull(x).replace(/[.,]\d{3}(?=\D*$)/, ''); + $('#tokenCount').textContent = tokens.list ? t('tokenCount', tokens.list.length) : ''; + $('#tokenExample').textContent = `curl -H "Authorization: Bearer lst_…" \\\n "${location.origin}/api/v1/logs?query=severity:err&since=15m&limit=1000"`; + if (!tokens.list) { + el.innerHTML = tokens.error ? `
${esc(tokens.error)}
` : ''; + return; + } + if (!tokens.list.length) { + el.innerHTML = `
${esc(t('tokenNone'))}
`; + return; + } + el.innerHTML = tokens.list.map((k) => { + const used = k.last_used ? t('tokenUsed', { d: day(k.last_used), ip: k.last_ip }) : t('tokenNever'); + return `
${esc(k.name)} ${esc(k.hint)}…
` + + `` + + `
${esc(t('tokenCreated', day(k.created)))} · ${esc(used)}
`; + }).join(''); +} + +async function loadTokens() { + $('#tokenNew').hidden = true; + try { + tokens.list = await api('/api/tokens'); + tokens.error = null; + } catch (e) { + tokens.list = null; + tokens.error = e.data?.code === 'read_only' ? '' : e.message; + } + renderTokens(); +} + +$('#tokenForm').addEventListener('submit', async (ev) => { + ev.preventDefault(); + const name = $('#tokenName').value.trim(); + if (!name) { $('#tokenName').focus(); return; } + try { + const res = await api('/api/tokens', { method: 'POST', body: { name } }); + $('#tokenName').value = ''; + await loadTokens(); + $('#tokenSecret').textContent = res.token; + $('#tokenNew').hidden = false; + } catch (e) { + toast(e.message); + } +}); + +$('#tokenCopy').addEventListener('click', () => copy($('#tokenSecret').textContent)); + +$('#tokenList').addEventListener('click', async (ev) => { + const b = ev.target.closest('[data-revoke]'); + if (!b) return; + const k = tokens.list?.find((x) => x.id === b.dataset.revoke); + if (!k || !confirm(t('tokenRevokeAsk', k.name))) return; + try { + await api('/api/tokens/' + encodeURIComponent(k.id), { method: 'DELETE' }); + toast(t('tokenRevoked')); + } catch (e) { + toast(e.message); + } + loadTokens(); +}); + /* ================= Tag management ================= */ const PALETTE = ['#6366f1', '#0ea5e9', '#14b8a6', '#a855f7', '#ec4899', '#eab308', '#64748b', '#f97316']; @@ -2828,6 +2934,7 @@ $('#settingsBtn').addEventListener('click', () => { loadDocker(); renderDbStats(); loadDbStats(); + loadTokens(); showSettingsTab(store.get('settingsTab', 'locale')); $('#settingsDlg').showModal(); }); @@ -3029,7 +3136,7 @@ fetch('/auth/me').then((res) => (res.ok ? res.json() : null)).then((me) => { } if (me.role === 'viewer') { document.body.classList.add('read-only'); - for (const p of document.querySelectorAll('[data-panel="filters"], [data-panel="sources"], [data-panel="data"]')) { + for (const p of document.querySelectorAll('[data-panel="filters"], [data-panel="sources"], [data-panel="data"], [data-panel="api"]')) { for (const s of p.querySelectorAll('.set-section')) s.inert = true; const note = document.createElement('p'); note.className = 'ro-note'; diff --git a/web/index.html b/web/index.html index d6a765b..dace1f8 100644 --- a/web/index.html +++ b/web/index.html @@ -197,6 +197,10 @@ Database +
+ + + diff --git a/web/style.css b/web/style.css index 90da40b..90d484b 100644 --- a/web/style.css +++ b/web/style.css @@ -912,3 +912,25 @@ body.login-page { .login-card input:focus { outline: 0; border-color: var(--accent); box-shadow: 0 0 0 3px var(--accent-soft); } .login-card .btn.primary { height: 38px; justify-content: center; margin-top: 6px; font-size: 14px; } .login-err { margin: 0; color: var(--sev-err); font-size: 13px; } + +/* API tokens (Settings > API) */ +.token-list { display: flex; flex-direction: column; gap: 4px; margin: 12px 0 10px; } +.token-row { + display: grid; grid-template-columns: minmax(0, 1fr) auto; gap: 2px 12px; align-items: center; + padding: 8px 10px; border-radius: 8px; background: var(--panel-2); font-size: 12.5px; +} +.token-row b { font-weight: 600; overflow-wrap: anywhere; } +.token-row code { font-family: var(--mono); font-size: 11.5px; color: var(--muted); } +.token-row .muted { grid-column: 1; font-size: 11.5px; } +.token-row .btn { grid-row: 1 / span 2; grid-column: 2; height: 28px; font-size: 12px; } +.token-empty { color: var(--muted); font-size: 12.5px; } +.token-form { display: flex; gap: 10px; align-items: center; } +.token-form .btn { height: 28px; font-size: 12px; } +.token-new { margin-top: 12px; padding: 10px 12px; border-radius: 8px; border: 1px solid var(--accent); background: var(--accent-soft); } +.token-secret { display: flex; gap: 10px; align-items: center; margin-top: 8px; } +.token-secret code { flex: 1; min-width: 0; font: 12px var(--mono); overflow-wrap: anywhere; user-select: all; } +.token-secret .btn { height: 28px; font-size: 12px; } +.token-example { + margin: 8px 0 0; padding: 8px 12px; border-radius: 8px; background: var(--panel-2); + font: 12px var(--mono); white-space: pre-wrap; overflow-wrap: anywhere; +} -- 2.54.0