Reading API /api/v1/logs with a stable cursor, and API tokens

GET /api/v1/logs?query=<LogsQL>&since=<cursor|time|duration>&limit=<n>
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) <noreply@anthropic.com>

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
+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.
+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` |

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 |

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)

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)
+	}
+}

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)
+	}
+}

docker-compose.yml

@@ -6,10 +6,10 @@ services:
     depends_on:
       - victorialogs
       - docker-proxy
-    ports:
+#    ports:
-      - "${SYSLOG_PORT:-5514}:5514/udp"
+#      - "${SYSLOG_PORT:-5514}:5514/udp"
-      - "${SYSLOG_PORT:-5514}:5514/tcp"
+#      - "${SYSLOG_PORT:-5514}:5514/tcp"
-      - "${HTTP_PORT:-8989}:8080"
+#      - "${HTTP_PORT:-8989}:8080"
     environment:
       VLOGS_URL: http://victorialogs:9428
       SYSLOG_PUBLIC_PORT: ${SYSLOG_PORT:-514}     # le port d'ecoute syslog par defaut (attention aux ports <1024)
@@ -60,12 +60,12 @@ services:
       - /run/log/journal:/host/run/log/journal:ro     # journal systemd volatile
       # audit Linux (auditd), source a activer dans Reglages > Sources ; le fichier doit etre
       # lisible par l'UID 10001 : log_group = adm dans /etc/audit/auditd.conf (voir le README)
-      # - /var/log/audit:/var/log/audit:ro
+      - /var/log/audit:/var/log/audit:ro
       # noms des utilisateurs et de l'hote pour les evenements d'audit (optionnel)
-      # - /etc/passwd:/host/etc/passwd:ro
+      - /etc/passwd:/host/etc/passwd:ro
-      # - /etc/hostname:/host/etc/hostname:ro
+      - /etc/hostname:/host/etc/hostname:ro
       # fichiers de logs d'autres programmes (Traefik, nginx...), a declarer dans Reglages > Sources > Fichiers de logs
-      # - /var/log/traefik:/logs/traefik:ro
+      - /home/traefik/log:/home/logs/traefik:ro
       # logo de la page de connexion, avec LOGIN_LOGO=/config/logo.png dans .env
       # - ./logo.png:/config/logo.png:ro
       # prereglages de tags personnalises, avec PRESETS_FILE=/config/presets.json dans .env
@@ -102,7 +102,12 @@ services:
       - /var/run/docker.sock:/var/run/docker.sock:ro
     labels:
       logstream.exclude: "true"      # pour exclure les logs propres de logstream
+    networks:
+      - traefik-net
 
+networks:
+  traefik-net:
+    external: true
 volumes:
   logstream-data:
   vlogs-data:

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=<LogsQL filter>&since=<cursor|time|duration>&limit=<n>&lag=<duration>
+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
+}

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))
+	}
+}

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,

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(),

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 ? `<div class="token-empty bad">${esc(tokens.error)}</div>` : '';
+    return;
+  }
+  if (!tokens.list.length) {
+    el.innerHTML = `<div class="token-empty">${esc(t('tokenNone'))}</div>`;
+    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 `<div class="token-row"><div><b>${esc(k.name)}</b> <code>${esc(k.hint)}…</code></div>`
+      + `<button class="btn danger" type="button" data-revoke="${esc(k.id)}">${esc(t('tokenRevoke'))}</button>`
+      + `<div class="muted">${esc(t('tokenCreated', day(k.created)))} · ${esc(used)}</div></div>`;
+  }).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';

web/index.html

@@ -197,6 +197,10 @@
             <svg viewBox="0 0 24 24"><ellipse cx="12" cy="5" rx="8" ry="3"/><path d="M4 5v14c0 1.7 3.6 3 8 3s8-1.3 8-3V5M4 12c0 1.7 3.6 3 8 3s8-1.3 8-3"/></svg>
             <span data-i18n="tabData">Database</span>
           </button>
+          <button type="button" role="tab" data-tab="api">
+            <svg viewBox="0 0 24 24"><circle cx="7.5" cy="15.5" r="4.5"/><path d="m10.7 12.3 9.3-9.3M16 7l3 3M18 5l2 2"/></svg>
+            <span data-i18n="tabApi">API</span>
+          </button>
         </nav>
         <div id="setActions" class="set-actions">
           <button id="setSave" class="btn save" type="button" disabled>
@@ -446,6 +450,34 @@
               </div>
             </section>
           </div>
+
+          <!-- API tokens -->
+          <div class="set-panel wide" role="tabpanel" data-panel="api" hidden>
+            <section class="set-section">
+              <div class="sec-head">
+                <h3 data-i18n="tokensTitle">API tokens</h3>
+                <span id="tokenCount" class="muted small"></span>
+              </div>
+              <p class="muted small" data-i18n="tokensHelp"></p>
+              <div id="tokenList" class="token-list"></div>
+              <form id="tokenForm" class="token-form">
+                <input id="tokenName" type="text" class="dk-filter" maxlength="80" autocomplete="off" spellcheck="false" data-i18n-ph="tokenNamePh" data-i18n-aria="tokenNamePh">
+                <button id="tokenCreate" class="btn" type="submit" data-i18n="tokenCreate">Create a token</button>
+              </form>
+              <div id="tokenNew" class="token-new" hidden>
+                <p class="small" data-i18n="tokenOnce"></p>
+                <div class="token-secret">
+                  <code id="tokenSecret"></code>
+                  <button id="tokenCopy" class="btn" type="button" data-i18n="tokenCopy">Copy</button>
+                </div>
+              </div>
+            </section>
+            <section class="set-section">
+              <h3 data-i18n="tokensUse">Usage</h3>
+              <p class="muted small" data-i18n="tokensUseHelp"></p>
+              <pre id="tokenExample" class="token-example"></pre>
+            </section>
+          </div>
         </div>
       </div>
     </div>

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;
+}