diff --git a/.env.example b/.env.example
index b5d5437..ea3268e 100644
--- a/.env.example
+++ b/.env.example
@@ -60,3 +60,10 @@ HOST_LOGS_BACKFILL=1h
# or run the container as root with the SYSLOG capability ("user" and "cap_add" in docker-compose.yml).
# History read when the source is turned on (e.g. 30m, 1h, 24h; 0 = only new messages)
DMESG_BACKFILL=1h
+# Linux audit log of the host (auditd, enable it in Settings > Sources). Uncomment the
+# /var/log/audit volume in docker-compose.yml and let the container read the file:
+# log_group = adm in /etc/audit/auditd.conf, and chgrp adm + chmod 750 on /var/log/audit.
+# History read when the source is turned on (e.g. 30m, 1h, 24h; 0 = only new events)
+AUDIT_BACKFILL=1h
+# Host name of the audit events (empty: node= of the records, else /host/etc/hostname, else localhost)
+AUDIT_HOST=
diff --git a/README.fr.md b/README.fr.md
index 2b28650..8d32912 100644
--- a/README.fr.md
+++ b/README.fr.md
@@ -240,6 +240,71 @@ défaut** : activez-la dans **Paramètres > Sources > Messages du noyau (dmesg)*
buffer est lu depuis `DMESG_BACKFILL`. Les messages écrasés dans le buffer du noyau avant
d'être lus sont comptés dans Paramètres.
+## Audit Linux (auditd)
+
+Logstream peut suivre le journal du démon d'audit Linux (`/var/log/audit/audit.log`) :
+connexions, `sudo`, fichiers surveillés, appels système visés par vos règles `auditctl`… La
+source est **désactivée par défaut** : activez-la dans **Réglages > Sources > Audit Linux
+(auditd)**, où le chemin du fichier peut aussi être changé (enregistré dans `/data/audit.json` ;
+`AUDIT_LOG` fixe le chemin par défaut).
+
+- **Montage** : décommentez `/var/log/audit:/var/log/audit:ro` dans `docker-compose.yml`, puis
+ `docker compose up -d`. (`/var/log` est déjà monté sous `/host` pour les logs système de
+ l'hôte : `/host/var/log/audit/audit.log` fonctionne donc aussi comme chemin, sans nouveau
+ volume.)
+- **Droits** (côté hôte) : auditd écrit son journal en `root`, mode `0600`, dans un répertoire
+ `0700`, et Logstream tourne avec l'UID `10001`. Le plus fiable est le groupe que le conteneur
+ reçoit déjà pour les logs de l'hôte (`HOST_LOGS_GID`, `adm` par défaut) :
+ ```bash
+ sudo sed -i 's/^log_group.*/log_group = adm/' /etc/audit/auditd.conf
+ sudo chgrp adm /var/log/audit && sudo chmod 750 /var/log/audit
+ sudo service auditd restart # systemctl refuse de redémarrer auditd sur certaines distributions
+ ```
+ auditd crée alors chaque fichier, rotations comprises, en mode `0640` avec le groupe `adm`.
+ Une ACL fonctionne aussi
+ (`sudo setfacl -m u:10001:rx /var/log/audit && sudo setfacl -m u:10001:r /var/log/audit/audit.log`)
+ mais doit être réappliquée à chaque nouveau fichier après une rotation. Réglages > Sources
+ indique si le fichier est absent ou l'accès refusé ; il est réessayé chaque seconde.
+- **Événements** : les enregistrements d'un même événement (`SYSCALL`, `EXECVE`, `CWD`, `PATH`,
+ `PROCTITLE`…) partagent le même `msg=audit(heure:numéro)` et deviennent **un seul** log, dès
+ qu'aucun enregistrement n'est arrivé depuis 2 secondes. Le message est un résumé — qui, quoi,
+ résultat — comme `cedric (as root): exec apt install nginx [exec] → success` ou
+ `bob: PAM:authentication via /usr/sbin/sshd from 203.0.113.9 on ssh → failed` ; les
+ enregistrements bruts sont gardés dans `audit_raw` et affichés sous le résumé (repliés en
+ densité compacte tant que la ligne n'est pas ouverte). Les valeurs hexadécimales
+ (`proctitle`, `cmd`, arguments avec espaces) sont décodées.
+- **Champs** : `audit_type` (type du premier enregistrement, aussi utilisé comme **app**),
+ `audit_serial`, `auid` (identifiant de l'utilisateur connecté), `user` (son nom, ou l'uid s'il
+ ne peut pas être résolu), `exe`, `res` (`success` / `failed`, d'après `res=` ou `success=`) et
+ `key` (clé de la règle). La sévérité est `warning` en cas d'échec, `info` sinon ; facility
+ `authpriv` ; source `audit`. Dans la liste, les lignes d'audit ont un liseré gauche violet,
+ un fond légèrement teinté et un badge **AUDIT** devant l'app ; le filtre **Source** permet de
+ les afficher seules.
+- **Noms** : `user` vient du format enrichi (`log_format = ENRICHED`, le défaut des versions
+ récentes d'auditd), sinon du `/etc/passwd` de l'hôte s'il est monté en `/host/etc/passwd`.
+ **host** est le `node=` des enregistrements (`name_format = hostname` dans `auditd.conf`),
+ sinon `AUDIT_HOST`, sinon `/host/etc/hostname`, sinon `localhost`.
+- À l'activation, la dernière heure est lue d'abord (`AUDIT_BACKFILL`, `0` = seulement les
+ nouveaux événements). L'inode et une position avant laquelle tout événement a été envoyé sont
+ enregistrés dans `/data/audit-state.json` : après un redémarrage, la lecture reprend là ;
+ après une rotation par auditd, la fin de `audit.log.1` est lue avant le nouveau fichier.
+
+## Logs internes de LogStream
+
+Logstream enregistre ses propres événements comme des logs, avec l'**hôte** `logstream`,
+l'**app** du composant (`store`, `syslog`, `docker`, `host`, `dmesg`, `audit`, `main`) et la
+source `logstream` : VictoriaLogs injoignable puis de retour, lots perdus, lots refusés depuis
+le tampon disque, erreurs du listener syslog, API Docker ou flux d'un conteneur décrochés,
+sources qui ne peuvent pas lire, démarrage et arrêt. Ils sont actifs par défaut à partir du
+niveau **warning** ; **Réglages > Sources > Logs internes de LogStream** permet de les couper
+ou de changer le niveau minimal (enregistré dans `/data/selflog.json`). Un même événement
+(composant et message) est enregistré au plus une fois toutes les 30 secondes, avec le nombre
+de répétitions écartées ; les événements passent par une file bornée et n'attendent jamais le
+stockage : un VictoriaLogs injoignable ne peut ni les accumuler ni les faire boucler. Tout
+continue d'aller dans `docker compose logs logstream` comme avant. Dans la liste, ces lignes
+ont un liseré gauche turquoise et un badge **LS** devant l'app ; le filtre **Source** permet de
+les afficher seules.
+
## Export CSV
Le bouton **Exporter** (à côté du nombre de logs) télécharge tous les logs stockés qui
@@ -435,6 +500,9 @@ résolutions.
| `HOST_LOGS_ROOT` | `/host` | emplacement de montage des répertoires de l'hôte |
| `DMESG_BACKFILL` | `1h` | historique lu à l'activation de la source « messages du noyau » |
| `DMESG_DEVICE` | `/dev/kmsg` | périphérique du log noyau lu par la source « messages du noyau » |
+| `AUDIT_LOG` | `/var/log/audit/audit.log` | chemin par défaut du journal d'audit Linux (modifiable dans Réglages > Sources) |
+| `AUDIT_BACKFILL` | `1h` | historique lu à l'activation de la source « audit Linux » |
+| `AUDIT_HOST` | vide | nom d'hôte des événements d'audit quand les enregistrements n'ont pas de `node=` |
| `TZ` | `Europe/Paris` | fuseau horaire des horodatages RFC 3164 (qui n'en portent pas) |
| `BATCH_SIZE`, `FLUSH_MS`, `QUEUE_SIZE` | `1000`, `1000`, `100000` | réglage de l'ingestion |
| `SPOOL_MAX_MB` | `1024` | taille du tampon disque des lots refusés par VictoriaLogs (`0` = pas de tampon : 15 s de tentatives, puis perte) |
@@ -496,6 +564,8 @@ Pour mettre à jour l'une d'elles :
| `syslogserver.go` | écoutes syslog ouvertes et fermées depuis Paramètres > Sources |
| `hostlogs.go`, `journal.go` | logs système de l'hôte : lecteur du journal systemd (sans `journalctl`) et suivi de `/var/log` |
| `dmesg.go` | messages du noyau : lecture de `/dev/kmsg`, analyse des enregistrements, position par identifiant de démarrage et numéro de séquence |
+| `audit.go` | audit Linux : suivi de `audit.log` (rotation, position enregistrée), analyse des enregistrements, regroupement par événement, résumé |
+| `selflog.go` | logs internes de LogStream : `logEvent`, niveau minimal, limitation des répétitions, file bornée |
| `tags.go` | stockage des tags de couleur |
| `presets.go`, `presets.json` | préréglages de tags (`/api/presets`), liste intégrée |
| `api.go` | routes HTTP `/api/*` |
diff --git a/README.md b/README.md
index 42d2309..46082e9 100644
--- a/README.md
+++ b/README.md
@@ -220,6 +220,65 @@ on in **Settings > Sources > Kernel messages (dmesg)** (saved in `/data/dmesg.js
`DMESG_BACKFILL` ago. Messages overwritten in the kernel buffer before being read are counted
in Settings.
+## Linux audit (auditd)
+
+Logstream can follow the log of the Linux audit daemon (`/var/log/audit/audit.log`): logins,
+`sudo`, watched files, syscalls matched by your `auditctl` rules… The source is **off by
+default**: turn it on in **Settings > Sources > Linux audit (auditd)**, where the file path can
+also be changed (saved in `/data/audit.json`; `AUDIT_LOG` sets the default path).
+
+- **Mount**: uncomment `/var/log/audit:/var/log/audit:ro` in `docker-compose.yml`, then
+ `docker compose up -d`. (`/var/log` is already mounted under `/host` for the host system
+ logs, so `/host/var/log/audit/audit.log` also works as the path, without a new volume.)
+- **Permissions** (host side): auditd writes its log as `root`, mode `0600`, in a `0700`
+ directory, and Logstream runs as UID `10001`. The reliable way is the group the container
+ already gets for the host logs (`HOST_LOGS_GID`, `adm` by default):
+ ```bash
+ sudo sed -i 's/^log_group.*/log_group = adm/' /etc/audit/auditd.conf
+ sudo chgrp adm /var/log/audit && sudo chmod 750 /var/log/audit
+ sudo service auditd restart # systemctl refuses to restart auditd on some distributions
+ ```
+ auditd then creates every rotated file with mode `0640` and group `adm`. An ACL also works
+ (`sudo setfacl -m u:10001:rx /var/log/audit && sudo setfacl -m u:10001:r /var/log/audit/audit.log`)
+ but must be applied again to each new file after a rotation. Settings > Sources says when the
+ file is missing or access is denied; it is tried again every second.
+- **Events**: the records of one event (`SYSCALL`, `EXECVE`, `CWD`, `PATH`, `PROCTITLE`…) share
+ the same `msg=audit(time:serial)` and become **one** log, once no record came for 2 seconds.
+ The message is a summary — who, what, result — such as
+ `cedric (as root): exec apt install nginx [exec] → success` or
+ `bob: PAM:authentication via /usr/sbin/sshd from 203.0.113.9 on ssh → failed`; the raw
+ records are kept in `audit_raw` and shown under the summary (folded in compact density until
+ the line is opened). Hexadecimal values (`proctitle`, `cmd`, arguments with spaces) are
+ decoded.
+- **Fields**: `audit_type` (type of the first record, also used as **app**), `audit_serial`,
+ `auid` (login user ID), `user` (its name, or the uid when it cannot be resolved), `exe`, `res`
+ (`success` / `failed`, from `res=` or `success=`) and `key` (rule key). Severity is `warning`
+ when the result is a failure, `info` otherwise; facility `authpriv`; source `audit`. In the
+ list, audit lines have a violet left edge, a light tint and an **AUDIT** badge before the app;
+ the **Source** filter shows them alone.
+- **Names**: `user` comes from the enriched format (`log_format = ENRICHED`, the default of
+ recent auditd versions), else from the host's `/etc/passwd` when it is mounted as
+ `/host/etc/passwd`. **host** is the `node=` of the records (`name_format = hostname` in
+ `auditd.conf`), else `AUDIT_HOST`, else `/host/etc/hostname`, else `localhost`.
+- When the source is turned on, the last hour is read first (`AUDIT_BACKFILL`, `0` = only new
+ events). The inode and an offset before which every event was sent are saved in
+ `/data/audit-state.json`: after a restart, reading resumes there; after a rotation by auditd,
+ the end of `audit.log.1` is read before the new file.
+
+## LogStream internal logs
+
+Logstream records its own events as logs, with **host** `logstream`, **app** the component
+(`store`, `syslog`, `docker`, `host`, `dmesg`, `audit`, `main`) and the `logstream` source:
+VictoriaLogs unreachable and back, lost batches, batches refused from the disk buffer, syslog
+listener errors, Docker API or container streams lost, sources that cannot read, start and
+stop. They are on by default from level **warning**; **Settings > Sources > LogStream internal
+logs** turns them off or changes the minimum level (saved in `/data/selflog.json`). The same
+event (component and message) is recorded at most once every 30 seconds, with the number of
+repeats left out; events go through a bounded queue and never wait for the storage, so an
+unreachable VictoriaLogs cannot make them pile up or loop. Everything still goes to
+`docker compose logs logstream` as before. In the list, these lines have a turquoise left edge
+and an **LS** badge before the app; the **Source** filter shows them alone.
+
## CSV export
The **Export** button (next to the log count) downloads every stored log matching the
@@ -398,6 +457,9 @@ are only known by your router or a local DNS (Pi-hole, AdGuard, Unbound…), set
| `HOST_LOGS_ROOT` | `/host` | where the host directories are mounted |
| `DMESG_BACKFILL` | `1h` | history read when the kernel messages source is turned on |
| `DMESG_DEVICE` | `/dev/kmsg` | kernel log device read by the kernel messages source |
+| `AUDIT_LOG` | `/var/log/audit/audit.log` | default path of the Linux audit log (can be changed in Settings > Sources) |
+| `AUDIT_BACKFILL` | `1h` | history read when the Linux audit source is turned on |
+| `AUDIT_HOST` | empty | host name of the audit events when the records carry no `node=` |
| `TZ` | `Europe/Paris` | time zone for RFC 3164 timestamps (which carry none) |
| `BATCH_SIZE`, `FLUSH_MS`, `QUEUE_SIZE` | `1000`, `1000`, `100000` | ingestion tuning |
| `SPOOL_MAX_MB` | `1024` | disk buffer size for batches VictoriaLogs could not take (`0` = no buffer: retried for 15 s, then dropped) |
@@ -457,6 +519,8 @@ To update one of them:
| `syslogserver.go` | syslog listeners opened and closed from Settings > Sources |
| `hostlogs.go`, `journal.go` | host system logs: systemd journal reader (no `journalctl`) and `/var/log` follower |
| `dmesg.go` | kernel messages: `/dev/kmsg` reader, record parsing, position by boot ID and sequence number |
+| `audit.go` | Linux audit: `audit.log` follower (rotation, saved offset), record parsing, grouping by event, summary |
+| `selflog.go` | LogStream internal logs: `logEvent`, minimum level, throttling, bounded queue |
| `tags.go` | color tag storage |
| `presets.go`, `presets.json` | color tag presets (`/api/presets`), built-in list |
| `api.go` | `/api/*` HTTP routes |
diff --git a/api.go b/api.go
index 6cd6303..92165ac 100644
--- a/api.go
+++ b/api.go
@@ -24,6 +24,8 @@ type API struct {
syslog *SyslogServer
host *HostLogs
dmesg *Dmesg
+ audit *Audit
+ selfLog *SelfLog
dbstats dbStatsCache
}
@@ -52,6 +54,50 @@ func (a *API) Routes(mux *http.ServeMux) {
mux.HandleFunc("PUT /api/hostlogs", a.hostLogsConfigure)
mux.HandleFunc("GET /api/dmesg", a.dmesgStatus)
mux.HandleFunc("PUT /api/dmesg", a.dmesgConfigure)
+ mux.HandleFunc("GET /api/audit", a.auditStatus)
+ mux.HandleFunc("PUT /api/audit", a.auditConfigure)
+ mux.HandleFunc("GET /api/selflog", a.selfLogStatus)
+ mux.HandleFunc("PUT /api/selflog", a.selfLogConfigure)
+}
+
+// GET /api/audit: state of the Linux audit source (Settings > Sources).
+func (a *API) auditStatus(w http.ResponseWriter, r *http.Request) {
+ writeJSON(w, http.StatusOK, a.audit.Status())
+}
+
+// PUT /api/audit {"enabled": bool, "path": "/var/log/audit/audit.log"}
+func (a *API) auditConfigure(w http.ResponseWriter, r *http.Request) {
+ var cfg auditConfig
+ if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4096)).Decode(&cfg); err != nil {
+ writeErr(w, http.StatusBadRequest, err)
+ return
+ }
+ if err := a.audit.Configure(cfg); err != nil {
+ writeErr(w, http.StatusInternalServerError, err)
+ return
+ }
+ log.Printf("audit source changed from %s: %+v", r.RemoteAddr, cfg)
+ writeJSON(w, http.StatusOK, a.audit.Status())
+}
+
+// GET /api/selflog: state of the LogStream internal logs (Settings > Sources).
+func (a *API) selfLogStatus(w http.ResponseWriter, r *http.Request) {
+ writeJSON(w, http.StatusOK, a.selfLog.Status())
+}
+
+// PUT /api/selflog {"enabled": bool, "min_level": "warning"}
+func (a *API) selfLogConfigure(w http.ResponseWriter, r *http.Request) {
+ var cfg selfLogConfig
+ if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4096)).Decode(&cfg); err != nil {
+ writeErr(w, http.StatusBadRequest, err)
+ return
+ }
+ if err := a.selfLog.Configure(cfg); err != nil {
+ writeErr(w, http.StatusInternalServerError, err)
+ return
+ }
+ log.Printf("internal logs changed from %s: %+v", r.RemoteAddr, cfg)
+ writeJSON(w, http.StatusOK, a.selfLog.Status())
}
// GET /api/dmesg: state of the kernel messages source (Settings > Sources).
diff --git a/audit.go b/audit.go
new file mode 100644
index 0000000..f469acd
--- /dev/null
+++ b/audit.go
@@ -0,0 +1,830 @@
+package main
+
+import (
+ "bufio"
+ "bytes"
+ "context"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "io"
+ "io/fs"
+ "os"
+ "path/filepath"
+ "sort"
+ "strconv"
+ "strings"
+ "sync"
+ "syscall"
+ "time"
+ "unicode/utf8"
+)
+
+// auditConfig is saved in /data/audit.json and edited in Settings > Sources.
+type auditConfig struct {
+ Enabled bool `json:"enabled"`
+ Path string `json:"path"` // empty: AUDIT_LOG (/var/log/audit/audit.log)
+}
+
+// auditPos is where reading resumes: the inode of the file and an offset
+// before which every event was sent.
+type auditPos struct {
+ Ino uint64 `json:"ino"`
+ Off int64 `json:"off"`
+}
+
+// Audit follows the log of the Linux audit daemon (auditd) and sends one
+// entry per audit event: the records of an event (SYSCALL, EXECVE, PATH…)
+// share the same msg=audit(time:serial) and are grouped together.
+type Audit struct {
+ defPath string
+ hostRoot string // to read the host's /etc/passwd and /etc/hostname when mounted
+ hostName string // AUDIT_HOST: host name of the entries when records have no node=
+ sink func(*Entry)
+ backfill time.Duration
+ cfgPath string
+ statePath string
+ wake chan struct{}
+
+ // Owned by the Run goroutine.
+ pos auditPos // checkpoint saved in /data
+ readOff int64 // where the next read starts (after the lines of open events)
+ known bool
+ dirty bool
+ notBefore time.Time
+ asm *auditAssembler
+ users *userNames
+
+ mu sync.Mutex
+ cfg auditConfig
+ reset bool
+ open bool
+ read uint64
+ errCode string
+ errDetail string
+}
+
+func NewAudit(defPath, hostRoot, hostName, dataDir string, backfill time.Duration, sink func(*Entry)) *Audit {
+ a := &Audit{
+ defPath: defPath,
+ hostRoot: hostRoot,
+ hostName: hostName,
+ sink: sink,
+ backfill: backfill,
+ cfgPath: filepath.Join(dataDir, "audit.json"),
+ statePath: filepath.Join(dataDir, "audit-state.json"),
+ wake: make(chan struct{}, 1),
+ asm: newAuditAssembler(),
+ users: &userNames{path: filepath.Join(hostRoot, "etc/passwd")},
+ }
+ if b, err := os.ReadFile(a.cfgPath); err == nil {
+ _ = json.Unmarshal(b, &a.cfg)
+ }
+ if b, err := os.ReadFile(a.statePath); err == nil && json.Unmarshal(b, &a.pos) == nil {
+ a.known, a.readOff = true, a.pos.Off
+ }
+ if a.hostName == "" {
+ a.hostName = readTrim(filepath.Join(hostRoot, "etc/hostname"))
+ }
+ return a
+}
+
+func (a *Audit) path() string {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ if a.cfg.Path != "" {
+ return a.cfg.Path
+ }
+ return a.defPath
+}
+
+// Run polls the file every second while the source is enabled.
+func (a *Audit) Run(ctx context.Context) {
+ tick := time.NewTicker(time.Second)
+ defer tick.Stop()
+ n := 0
+ for {
+ select {
+ case <-ctx.Done():
+ a.emit(a.asm.flushAll())
+ a.checkpoint()
+ a.saveState()
+ return
+ case <-tick.C:
+ case <-a.wake:
+ }
+ a.mu.Lock()
+ enabled, reset := a.cfg.Enabled, a.reset
+ a.reset = false
+ a.mu.Unlock()
+ if reset {
+ // Disabled then enabled again (or another file): start over from
+ // AUDIT_BACKFILL ago rather than reading everything in between.
+ a.asm = newAuditAssembler()
+ a.pos, a.readOff, a.known, a.dirty = auditPos{}, 0, false, true
+ }
+ if enabled {
+ a.poll(time.Now())
+ }
+ if n++; n%5 == 0 || reset {
+ a.saveState()
+ }
+ }
+}
+
+func (a *Audit) saveState() {
+ if !a.dirty {
+ return
+ }
+ a.dirty = false
+ if err := writeJSONFile(a.statePath, a.pos); err != nil {
+ logEvent("audit", 4, "saving position: %v", err)
+ }
+}
+
+func (a *Audit) fail(err error) {
+ code := "read"
+ switch {
+ case errors.Is(err, fs.ErrNotExist):
+ code = "not_found"
+ case errors.Is(err, fs.ErrPermission):
+ code = "permission"
+ }
+ a.mu.Lock()
+ changed := code != a.errCode
+ a.open, a.errCode, a.errDetail = false, code, err.Error()
+ a.mu.Unlock()
+ if changed {
+ logEvent("audit", 4, "cannot read the audit log: %v", err)
+ }
+}
+
+const maxAuditRead = 4 << 20 // per poll
+
+// poll reads the lines added since the previous one, follows rotations and
+// sends the events that are complete.
+func (a *Audit) poll(now time.Time) {
+ path := a.path()
+ fi, err := os.Stat(path)
+ if err != nil {
+ a.fail(err)
+ return
+ }
+ ino := inodeOf(fi)
+ if !a.known {
+ // First read: the last AUDIT_BACKFILL (0: only new events).
+ a.pos, a.readOff, a.known, a.dirty = auditPos{Ino: ino}, 0, true, true
+ a.notBefore = time.Time{}
+ if a.backfill <= 0 {
+ a.readOff = fi.Size()
+ } else {
+ a.notBefore = now.Add(-a.backfill)
+ }
+ }
+ if a.pos.Ino != ino {
+ // Rotated by auditd (audit.log -> audit.log.1): finish the old file,
+ // then read the new one from its start.
+ if old, err := os.Stat(path + ".1"); err == nil && inodeOf(old) == a.pos.Ino {
+ for {
+ off, more, err := a.readFrom(path+".1", a.readOff, now)
+ if err != nil || !more {
+ break
+ }
+ a.readOff = off
+ }
+ }
+ a.emit(a.asm.flushAll())
+ a.pos, a.readOff, a.dirty = auditPos{Ino: ino}, 0, true
+ } else if fi.Size() < a.readOff {
+ a.emit(a.asm.flushAll()) // truncated
+ a.readOff = 0
+ }
+ if fi.Size() > a.readOff {
+ off, _, err := a.readFrom(path, a.readOff, now)
+ if err != nil {
+ a.fail(err)
+ return
+ }
+ a.readOff = off
+ }
+ a.mu.Lock()
+ if !a.open {
+ a.open, a.errCode, a.errDetail = true, "", ""
+ }
+ a.mu.Unlock()
+ a.emit(a.asm.flush(now, auditCloseDelay))
+ a.checkpoint()
+}
+
+// checkpoint saves an offset before which every event is sent: the start of
+// the oldest event still open, or the read offset.
+func (a *Audit) checkpoint() {
+ off := a.readOff
+ if o, ok := a.asm.minOff(); ok && o < off {
+ off = o
+ }
+ if off != a.pos.Off {
+ a.pos.Off, a.dirty = off, true
+ }
+}
+
+// readFrom feeds the complete lines written after off to the assembler and
+// returns the new offset; more tells whether the read window was full.
+func (a *Audit) readFrom(path string, off int64, now time.Time) (int64, bool, error) {
+ f, err := os.Open(path)
+ if err != nil {
+ return off, false, err
+ }
+ defer f.Close()
+ buf, err := io.ReadAll(io.NewSectionReader(f, off, maxAuditRead))
+ if err != nil {
+ return off, false, err
+ }
+ full := len(buf) == maxAuditRead
+ end := bytes.LastIndexByte(buf, '\n')
+ if end < 0 {
+ if !full {
+ return off, false, nil // partial line: wait for its end
+ }
+ end = len(buf) - 1
+ }
+ start := off
+ for _, line := range bytes.Split(buf[:end], []byte{'\n'}) {
+ a.emit(a.asm.add(string(line), start, now))
+ start += int64(len(line)) + 1
+ }
+ return off + int64(end) + 1, full, nil
+}
+
+func (a *Audit) emit(evs []*auditEvent) {
+ for _, ev := range evs {
+ if !a.notBefore.IsZero() && ev.Time.Before(a.notBefore) {
+ continue
+ }
+ a.sink(auditEntry(ev, a.users, a.hostName, time.Now()))
+ a.mu.Lock()
+ a.read++
+ a.mu.Unlock()
+ }
+}
+
+func inodeOf(fi os.FileInfo) uint64 {
+ if st, ok := fi.Sys().(*syscall.Stat_t); ok {
+ return uint64(st.Ino)
+ }
+ return 0
+}
+
+// Configure saves and applies the configuration.
+func (a *Audit) Configure(cfg auditConfig) error {
+ cfg.Path = strings.TrimSpace(cfg.Path)
+ a.mu.Lock()
+ if (a.cfg.Enabled && !cfg.Enabled) || cfg.Path != a.cfg.Path {
+ a.reset = true
+ a.open, a.errCode, a.errDetail = false, "", ""
+ }
+ a.cfg = cfg
+ a.mu.Unlock()
+ select {
+ case a.wake <- struct{}{}:
+ default:
+ }
+ return writeJSONFile(a.cfgPath, cfg)
+}
+
+// Status is the state shown in Settings > Sources.
+func (a *Audit) Status() map[string]any {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ return map[string]any{
+ "enabled": a.cfg.Enabled,
+ "path": a.cfg.Path,
+ "defaultPath": a.defPath,
+ "open": a.open && a.cfg.Enabled,
+ "read": a.read,
+ "code": a.errCode,
+ "error": a.errDetail,
+ }
+}
+
+/* ---------- Records and events ---------- */
+
+// auditCloseDelay: an event is complete when no line of it came for this
+// long, or when the log has moved this far past its time (auditd does not
+// write the EOE record that ends multi-record events).
+const auditCloseDelay = 2 * time.Second
+
+type auditField struct {
+ k, v string
+ bare bool // written without quotes: may be hexadecimal
+}
+
+type auditRecord struct {
+ Type string
+ Fields []auditField
+ Raw string
+}
+
+type auditEvent struct {
+ ID string // "1700000000.123:456"
+ Serial string
+ Time time.Time
+ Node string
+ Records []auditRecord
+ off int64 // offset of its first line in the file
+ seen time.Time // when its last line was read
+ order int
+}
+
+// parseAuditLine decodes "[node=N ]type=T msg=audit(SEC.MS:SERIAL): k=v …".
+func parseAuditLine(line string) (rec auditRecord, id, node string, ok bool) {
+ line = strings.TrimRight(line, "\r")
+ rec.Raw = line
+ head, rest, found := strings.Cut(line, "): ")
+ if !found {
+ head, found = strings.CutSuffix(line, ")")
+ if !found {
+ return rec, "", "", false
+ }
+ }
+ pre, ts, found := strings.Cut(head, "msg=audit(")
+ if !found || ts == "" {
+ return rec, "", "", false
+ }
+ for _, f := range parseAuditKV(pre) {
+ switch f.k {
+ case "type":
+ rec.Type = f.v
+ case "node":
+ node = f.v
+ }
+ }
+ if rec.Type == "" {
+ return rec, "", "", false
+ }
+ // Enriched format (log_format=ENRICHED): resolved values after a GS byte.
+ main, enriched, _ := strings.Cut(rest, "\x1d")
+ rec.Fields = parseAuditKV(main)
+ for _, f := range parseAuditKV(enriched) {
+ rec.Fields = append(rec.Fields, auditField{k: "@" + f.k, v: f.v})
+ }
+ return rec, ts, node, true
+}
+
+// hexFields are the fields auditd writes in hexadecimal when they contain
+// spaces, quotes or control characters (and then without quotes).
+var hexFields = map[string]bool{
+ "proctitle": true, "cmd": true, "exe": true, "comm": true, "key": true, "acct": true,
+ "name": true, "cwd": true, "data": true, "path": true, "old": true, "new": true,
+}
+
+// parseAuditKV splits "k=v k2="v 2" msg='k3=v3 …'"; the fields of a quoted
+// msg='…' (user space records) follow the outer ones.
+func parseAuditKV(s string) []auditField {
+ var out []auditField
+ var inner []auditField
+ for i := 0; i < len(s); {
+ for i < len(s) && s[i] == ' ' {
+ i++
+ }
+ eq := strings.IndexByte(s[i:], '=')
+ if eq <= 0 {
+ break
+ }
+ k := s[i : i+eq]
+ if sp := strings.IndexByte(k, ' '); sp >= 0 { // word without value
+ i += sp + 1
+ continue
+ }
+ i += eq + 1
+ var v string
+ quoted := i < len(s) && (s[i] == '"' || s[i] == '\'')
+ if quoted {
+ q := s[i]
+ end := strings.IndexByte(s[i+1:], q)
+ if end < 0 {
+ end = len(s) - i - 1
+ }
+ v = s[i+1 : i+1+end]
+ i += end + 2
+ } else {
+ end := strings.IndexByte(s[i:], ' ')
+ if end < 0 {
+ end = len(s) - i
+ }
+ v = s[i : i+end]
+ i += end
+ }
+ if k == "msg" && quoted && strings.Contains(v, "=") {
+ inner = append(inner, parseAuditKV(v)...)
+ continue
+ }
+ if !quoted && hexFields[k] {
+ v = decodeHex(k, v)
+ }
+ if !quoted && (v == "(null)" || v == "?") {
+ v = ""
+ }
+ out = append(out, auditField{k: k, v: v, bare: !quoted})
+ }
+ return append(out, inner...)
+}
+
+// decodeHex decodes a hexadecimal value, or returns it unchanged.
+func decodeHex(k, v string) string {
+ if len(v) < 2 || len(v)%2 != 0 {
+ return v
+ }
+ b, err := hex.DecodeString(v)
+ if err != nil {
+ return v
+ }
+ if k == "proctitle" {
+ b = bytes.ReplaceAll(bytes.TrimRight(b, "\x00"), []byte{0}, []byte{' '})
+ }
+ if !utf8.Valid(b) {
+ return v
+ }
+ return string(b)
+}
+
+func isDigits(s string) bool {
+ for i := 0; i < len(s); i++ {
+ if s[i] < '0' || s[i] > '9' {
+ return false
+ }
+ }
+ return s != ""
+}
+
+// parseAuditTime decodes "1700000000.123:456" into a time and a serial.
+func parseAuditTime(id string) (time.Time, string) {
+ ts, serial, _ := strings.Cut(id, ":")
+ sec, frac, _ := strings.Cut(ts, ".")
+ s, err := strconv.ParseInt(sec, 10, 64)
+ if err != nil {
+ return time.Time{}, serial
+ }
+ ms, _ := strconv.ParseInt((frac + "000")[:3], 10, 64)
+ return time.Unix(s, ms*int64(time.Millisecond)), serial
+}
+
+// auditAssembler groups the records of each event.
+type auditAssembler struct {
+ open map[string]*auditEvent
+ newest time.Time // most recent event time read
+ n int
+}
+
+func newAuditAssembler() *auditAssembler {
+ return &auditAssembler{open: map[string]*auditEvent{}}
+}
+
+const maxOpenAudit = 2000
+
+// add takes one line read at offset off and returns the events it completes
+// (EOE record, or too many events open).
+func (a *auditAssembler) add(line string, off int64, now time.Time) []*auditEvent {
+ if strings.TrimSpace(line) == "" {
+ return nil
+ }
+ rec, id, node, ok := parseAuditLine(line)
+ if !ok {
+ return nil
+ }
+ ev := a.open[id]
+ if rec.Type == "EOE" {
+ if ev == nil {
+ return nil
+ }
+ delete(a.open, id)
+ return []*auditEvent{ev}
+ }
+ if ev == nil {
+ t, serial := parseAuditTime(id)
+ a.n++
+ ev = &auditEvent{ID: id, Serial: serial, Time: t, Node: node, off: off, order: a.n}
+ a.open[id] = ev
+ if t.After(a.newest) {
+ a.newest = t
+ }
+ }
+ ev.Records = append(ev.Records, rec)
+ ev.seen = now
+ if len(a.open) > maxOpenAudit {
+ return a.closeOldest(len(a.open) - maxOpenAudit/2)
+ }
+ return nil
+}
+
+// flush returns the events with no new line for delay, or that the log has
+// left delay behind, in the order they started.
+func (a *auditAssembler) flush(now time.Time, delay time.Duration) []*auditEvent {
+ var out []*auditEvent
+ for id, ev := range a.open {
+ if now.Sub(ev.seen) >= delay || a.newest.Sub(ev.Time) >= delay {
+ out = append(out, ev)
+ delete(a.open, id)
+ }
+ }
+ sortEvents(out)
+ return out
+}
+
+func (a *auditAssembler) flushAll() []*auditEvent {
+ return a.closeOldest(len(a.open))
+}
+
+func (a *auditAssembler) closeOldest(n int) []*auditEvent {
+ all := make([]*auditEvent, 0, len(a.open))
+ for _, ev := range a.open {
+ all = append(all, ev)
+ }
+ sortEvents(all)
+ all = all[:n]
+ for _, ev := range all {
+ delete(a.open, ev.ID)
+ }
+ return all
+}
+
+func sortEvents(evs []*auditEvent) {
+ sort.Slice(evs, func(i, j int) bool { return evs[i].order < evs[j].order })
+}
+
+func (a *auditAssembler) minOff() (int64, bool) {
+ var m int64
+ ok := false
+ for _, ev := range a.open {
+ if !ok || ev.off < m {
+ m, ok = ev.off, true
+ }
+ }
+ return m, ok
+}
+
+/* ---------- Entries ---------- */
+
+// get returns the first non-empty value of a field across the records.
+func (ev *auditEvent) get(k string) string {
+ for _, r := range ev.Records {
+ for _, f := range r.Fields {
+ if f.k == k && f.v != "" {
+ return f.v
+ }
+ }
+ }
+ return ""
+}
+
+func (ev *auditEvent) record(typ string) *auditRecord {
+ for i := range ev.Records {
+ if ev.Records[i].Type == typ {
+ return &ev.Records[i]
+ }
+ }
+ return nil
+}
+
+// result is "success", "failed" or "" from res= or success=.
+func (ev *auditEvent) result() string {
+ v := strings.ToLower(firstNonEmpty(ev.get("res"), ev.get("success"), ev.get("result")))
+ switch v {
+ case "success", "yes", "1":
+ return "success"
+ case "failed", "fail", "failure", "no", "0":
+ return "failed"
+ }
+ return ""
+}
+
+const auditUnset = "4294967295"
+
+// userOf resolves a uid: enriched name, then the host's /etc/passwd, then the number.
+func (ev *auditEvent) userOf(field string, users *userNames) string {
+ uid := ev.get(field)
+ if uid == "" || uid == auditUnset || uid == "-1" {
+ return ""
+ }
+ if n := strings.Trim(ev.get("@"+strings.ToUpper(field)), `"`); n != "" && n != "unset" && n != uid {
+ return n
+ }
+ if n := users.name(uid); n != "" {
+ return n
+ }
+ return uid
+}
+
+// auditEntry turns an event into an Entry: app is the type of its first record.
+func auditEntry(ev *auditEvent, users *userNames, hostName string, now time.Time) *Entry {
+ typ := "UNKNOWN"
+ if len(ev.Records) > 0 {
+ typ = ev.Records[0].Type
+ }
+ res := ev.result()
+ auid := ev.get("auid")
+ user := ev.userOf("auid", users)
+ if user == "" {
+ user = ev.userOf("uid", users)
+ }
+ sev := 6
+ if res == "failed" {
+ sev = 4
+ }
+ raw := make([]string, len(ev.Records))
+ for i, r := range ev.Records {
+ raw[i] = r.Raw
+ }
+ host := firstNonEmpty(ev.Node, hostName, "localhost")
+ e := &Entry{
+ Time: ev.Time,
+ Received: now,
+ Host: host,
+ App: typ,
+ ProcID: ev.get("pid"),
+ Facility: "authpriv",
+ Severity: severityNames[sev],
+ SevNum: sev,
+ Message: auditSummary(ev, typ, user, users, res),
+ Proto: "file",
+ SourceType: "audit",
+ Extra: map[string]string{
+ "audit_type": typ,
+ "audit_serial": ev.Serial,
+ "auid": auid,
+ "user": user,
+ "exe": ev.get("exe"),
+ "res": res,
+ "key": ev.get("key"),
+ "audit_raw": strings.Join(raw, "\n"),
+ },
+ Wait: true,
+ }
+ if e.Time.IsZero() {
+ e.Time = now
+ }
+ return e
+}
+
+// auditSummary writes "who: what [key] → result".
+func auditSummary(ev *auditEvent, typ, user string, users *userNames, res string) string {
+ who := firstNonEmpty(user, "?")
+ // Syscalls: the login user and, when different, the user the process ran
+ // as (sudo, su). In user space records, uid is the daemon's (sshd: root).
+ if uid := ev.userOf("uid", users); ev.get("syscall") != "" && uid != "" && user != "" && uid != user {
+ who += " (as " + uid + ")"
+ }
+ var what string
+ exe := ev.get("exe")
+ switch {
+ case typ == "USER_CMD":
+ what = "sudo " + firstNonEmpty(ev.get("cmd"), "?")
+ if cwd := ev.get("cwd"); cwd != "" {
+ what += " (in " + cwd + ")"
+ }
+ case strings.HasPrefix(typ, "USER_") || strings.HasPrefix(typ, "CRED_") || typ == "LOGIN" ||
+ typ == "ADD_USER" || typ == "DEL_USER" || typ == "ADD_GROUP" || typ == "DEL_GROUP":
+ if acct := ev.get("acct"); acct != "" {
+ who = acct
+ }
+ what = firstNonEmpty(ev.get("op"), strings.ToLower(typ))
+ if exe != "" {
+ what += " via " + exe
+ }
+ if addr := firstNonEmpty(ev.get("hostname"), ev.get("addr")); addr != "" {
+ what += " from " + addr
+ }
+ if term := ev.get("terminal"); term != "" && term != "?" {
+ what += " on " + term
+ }
+ case ev.record("EXECVE") != nil:
+ what = "exec " + execLine(ev.record("EXECVE"))
+ case ev.get("syscall") != "":
+ what = "syscall " + firstNonEmpty(ev.get("@SYSCALL"), ev.get("syscall"))
+ if name := pathName(ev); name != "" {
+ what += " " + name
+ }
+ if prog := firstNonEmpty(exe, ev.get("comm")); prog != "" {
+ what += " by " + prog
+ }
+ default:
+ what = strings.ToLower(typ)
+ if op := ev.get("op"); op != "" {
+ what += " " + op
+ }
+ if exe != "" {
+ what += " by " + exe
+ }
+ }
+ if key := ev.get("key"); key != "" {
+ what += " [" + key + "]"
+ }
+ if res != "" {
+ what += " → " + res
+ }
+ return who + ": " + what
+}
+
+// execLine joins the arguments of an EXECVE record (a0, a1…).
+func execLine(r *auditRecord) string {
+ args := map[int]string{}
+ argc := 0
+ for _, f := range r.Fields {
+ if f.k == "argc" {
+ argc, _ = strconv.Atoi(f.v)
+ } else if len(f.k) > 1 && f.k[0] == 'a' && isDigits(f.k[1:]) {
+ n, _ := strconv.Atoi(f.k[1:])
+ if f.bare {
+ args[n] = decodeHex(f.k, f.v) // arguments with spaces or quotes
+ } else {
+ args[n] = f.v
+ }
+ }
+ }
+ if argc == 0 {
+ argc = len(args)
+ }
+ parts := make([]string, 0, argc)
+ for i := 0; i < argc; i++ {
+ if v, ok := args[i]; ok {
+ parts = append(parts, v)
+ }
+ }
+ s := strings.Join(parts, " ")
+ if len(s) > 300 {
+ s = s[:300] + "…"
+ }
+ return s
+}
+
+// pathName is the file a syscall worked on: the first PATH record that is not
+// its parent directory.
+func pathName(ev *auditEvent) string {
+ var first string
+ for _, r := range ev.Records {
+ if r.Type != "PATH" {
+ continue
+ }
+ var name, nametype string
+ for _, f := range r.Fields {
+ switch f.k {
+ case "name":
+ name = f.v
+ case "nametype":
+ nametype = f.v
+ }
+ }
+ if name == "" {
+ continue
+ }
+ if nametype != "PARENT" {
+ return name
+ }
+ if first == "" {
+ first = name
+ }
+ }
+ return first
+}
+
+/* ---------- User names ---------- */
+
+// userNames reads uid -> name from the host's /etc/passwd when it is mounted,
+// at most once a minute.
+type userNames struct {
+ path string
+ names map[string]string
+ loaded time.Time
+ modTime time.Time
+}
+
+func (u *userNames) name(uid string) string {
+ if u == nil {
+ return ""
+ }
+ if time.Since(u.loaded) > time.Minute {
+ u.loaded = time.Now()
+ if fi, err := os.Stat(u.path); err == nil && !fi.ModTime().Equal(u.modTime) {
+ u.modTime = fi.ModTime()
+ u.names = readPasswd(u.path)
+ }
+ }
+ return u.names[uid]
+}
+
+func readPasswd(path string) map[string]string {
+ f, err := os.Open(path)
+ if err != nil {
+ return nil
+ }
+ defer f.Close()
+ names := map[string]string{}
+ sc := bufio.NewScanner(f)
+ for sc.Scan() {
+ p := strings.Split(sc.Text(), ":")
+ if len(p) >= 3 && p[0] != "" && !strings.HasPrefix(p[0], "#") {
+ names[p[2]] = p[0]
+ }
+ }
+ return names
+}
diff --git a/audit_test.go b/audit_test.go
new file mode 100644
index 0000000..9ca3afc
--- /dev/null
+++ b/audit_test.go
@@ -0,0 +1,356 @@
+package main
+
+import (
+ "context"
+ "encoding/hex"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+)
+
+func hexOf(s string) string { return strings.ToUpper(hex.EncodeToString([]byte(s))) }
+
+// Lines of real audit.log events (trimmed).
+var (
+ auditExec = []string{
+ `type=SYSCALL msg=audit(1759572000.123:4242): arch=c000003e syscall=59 success=yes exit=0 a0=4142 a1=55d2 a2=55d3 a3=0 items=2 ppid=1000 pid=1234 auid=1000 uid=0 gid=0 euid=0 tty=pts0 ses=3 comm="apt" exe="/usr/bin/apt" subj=unconfined key="exec"`,
+ `type=EXECVE msg=audit(1759572000.123:4242): argc=3 a0="apt" a1="install" a2=` + hexOf(`my "pkg"`),
+ `type=CWD msg=audit(1759572000.123:4242): cwd="/root"`,
+ `type=PATH msg=audit(1759572000.123:4242): item=0 name="/usr/bin/apt" inode=1 dev=fd:00 mode=0100755 ouid=0 ogid=0 nametype=NORMAL`,
+ `type=PROCTITLE msg=audit(1759572000.123:4242): proctitle=` + hexOf("apt\x00install\x00my \"pkg\""),
+ }
+ auditSudo = `type=USER_CMD msg=audit(1759572001.500:4250): pid=2000 uid=1000 auid=1000 ses=3 subj=unconfined msg='cwd="/home/cedric" cmd=` + hexOf("systemctl restart nginx") + ` exe="/usr/bin/sudo" terminal=pts/0 res=success'`
+ auditSSH = `type=USER_AUTH msg=audit(1759572002.000:4260): pid=3000 uid=0 auid=4294967295 ses=4294967295 subj=unconfined msg='op=PAM:authentication grantors=? acct="bob" exe="/usr/sbin/sshd" hostname=203.0.113.9 addr=203.0.113.9 terminal=ssh res=failed'`
+ auditOpen = []string{
+ "node=web1 type=SYSCALL msg=audit(1759572003.250:4270): arch=c000003e syscall=257 success=no exit=-13 items=1 pid=4000 auid=1001 uid=1001 comm=\"cat\" exe=\"/usr/bin/cat\" key=\"shadow\"\x1dARCH=x86_64 SYSCALL=openat AUID=\"alice\" UID=\"alice\"",
+ `node=web1 type=PATH msg=audit(1759572003.250:4270): item=0 name="/etc/shadow" inode=7 nametype=NORMAL`,
+ }
+)
+
+func TestParseAuditLine(t *testing.T) {
+ rec, id, node, ok := parseAuditLine(auditOpen[0])
+ if !ok || rec.Type != "SYSCALL" || id != "1759572003.250:4270" || node != "web1" {
+ t.Fatalf("got %+v %q %q %v", rec, id, node, ok)
+ }
+ ev := &auditEvent{Records: []auditRecord{rec}}
+ for k, want := range map[string]string{"syscall": "257", "exe": "/usr/bin/cat", "key": "shadow", "@SYSCALL": "openat", "@AUID": "alice"} {
+ if got := ev.get(k); got != want {
+ t.Errorf("%s = %q, want %q", k, got, want)
+ }
+ }
+ for _, bad := range []string{"", "garbage", "type=X no header", "msg=audit(1.2:3): no type"} {
+ if _, _, _, ok := parseAuditLine(bad); ok {
+ t.Errorf("%q parsed", bad)
+ }
+ }
+}
+
+func TestParseAuditKV(t *testing.T) {
+ tests := []struct {
+ in string
+ want map[string]string
+ }{
+ {`a=1 b="two words" c=`, map[string]string{"a": "1", "b": "two words", "c": ""}},
+ {`pid=1 msg='op=login acct="bob" res=success'`, map[string]string{"pid": "1", "op": "login", "acct": "bob", "res": "success"}},
+ {`comm=` + hexOf("my prog") + ` exe="/bin/x"`, map[string]string{"comm": "my prog", "exe": "/bin/x"}},
+ {`key=(null) hostname=? terminal=?`, map[string]string{"key": "", "hostname": "", "terminal": ""}},
+ {`a0=4142 name=ABC`, map[string]string{"a0": "4142", "name": "ABC"}}, // not hexadecimal fields, or odd length
+ {`avc: denied { read } for pid=12 comm="x"`, map[string]string{"pid": "12", "comm": "x"}},
+ }
+ for _, tt := range tests {
+ got := map[string]string{}
+ for _, f := range parseAuditKV(tt.in) {
+ got[f.k] = f.v
+ }
+ for k, v := range tt.want {
+ if got[k] != v {
+ t.Errorf("%q: %s = %q, want %q (all: %v)", tt.in, k, got[k], v, got)
+ }
+ }
+ }
+}
+
+func TestAuditAssembler(t *testing.T) {
+ now := time.Unix(1759572010, 0)
+ a := newAuditAssembler()
+ // Two interleaved events, then an EOE that closes the first one.
+ lines := []string{auditExec[0], auditSudo, auditExec[1], auditExec[2]}
+ off := int64(0)
+ for _, l := range lines {
+ if out := a.add(l, off, now); out != nil {
+ t.Fatalf("closed early: %v", out)
+ }
+ off += int64(len(l)) + 1
+ }
+ if len(a.open) != 2 {
+ t.Fatalf("%d open events", len(a.open))
+ }
+ if o, ok := a.minOff(); !ok || o != 0 {
+ t.Fatalf("minOff %d %v", o, ok)
+ }
+ out := a.add(`type=EOE msg=audit(1759572000.123:4242): `, off, now)
+ if len(out) != 1 || out[0].Serial != "4242" || len(out[0].Records) != 3 {
+ t.Fatalf("EOE: %+v", out)
+ }
+ if o, _ := a.minOff(); o != int64(len(lines[0])+1) {
+ t.Fatalf("minOff after EOE: %d", o)
+ }
+ // An EOE of an unknown event is ignored.
+ if out := a.add(`type=EOE msg=audit(1.0:1): `, 0, now); out != nil {
+ t.Fatal(out)
+ }
+ // The sudo event is closed by time.
+ if out := a.flush(now.Add(time.Second), 2*time.Second); len(out) != 0 {
+ t.Fatalf("closed too early: %v", out)
+ }
+ out = a.flush(now.Add(3*time.Second), 2*time.Second)
+ if len(out) != 1 || out[0].Records[0].Type != "USER_CMD" {
+ t.Fatalf("flush: %+v", out)
+ }
+ // Events the log has left behind are closed without waiting.
+ a.add(auditSSH, 0, now)
+ a.add(auditOpen[0], 0, now)
+ out = a.flush(now, 1*time.Second)
+ if len(out) != 1 || out[0].Serial != "4260" || len(a.open) != 1 {
+ t.Fatalf("newest: %+v, open %d", out, len(a.open))
+ }
+ if out = a.flushAll(); len(out) != 1 || out[0].Serial != "4270" || len(a.open) != 0 {
+ t.Fatalf("flushAll: %+v", out)
+ }
+}
+
+func auditEv(t *testing.T, lines ...string) *auditEvent {
+ t.Helper()
+ a := newAuditAssembler()
+ for _, l := range lines {
+ a.add(l, 0, time.Time{})
+ }
+ out := a.flushAll()
+ if len(out) != 1 {
+ t.Fatalf("%d events", len(out))
+ }
+ return out[0]
+}
+
+func TestAuditEntry(t *testing.T) {
+ dir := t.TempDir()
+ passwd := filepath.Join(dir, "passwd")
+ if err := os.WriteFile(passwd, []byte("root:x:0:0::/root:/bin/sh\ncedric:x:1000:1000::/home/cedric:/bin/bash\n"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ now := time.Unix(1759572100, 0)
+ tests := []struct {
+ name string
+ lines []string
+ host string
+ app string
+ sev string
+ msg string
+ fields map[string]string
+ }{
+ {"exec", auditExec, "srv", "SYSCALL", "info",
+ `cedric (as root): exec apt install my "pkg" [exec] → success`,
+ map[string]string{"audit_type": "SYSCALL", "audit_serial": "4242", "auid": "1000", "user": "cedric", "exe": "/usr/bin/apt", "res": "success", "key": "exec"}},
+ {"sudo", []string{auditSudo}, "srv", "USER_CMD", "info",
+ "cedric: sudo systemctl restart nginx (in /home/cedric) → success",
+ map[string]string{"user": "cedric", "exe": "/usr/bin/sudo", "res": "success", "key": ""}},
+ {"ssh failed", []string{auditSSH}, "srv", "USER_AUTH", "warning",
+ "bob: PAM:authentication via /usr/sbin/sshd from 203.0.113.9 on ssh → failed",
+ map[string]string{"auid": "4294967295", "user": "root", "res": "failed"}},
+ {"enriched, node", auditOpen, "web1", "SYSCALL", "warning",
+ "alice: syscall openat /etc/shadow by /usr/bin/cat [shadow] → failed",
+ map[string]string{"user": "alice", "auid": "1001", "res": "failed", "key": "shadow"}},
+ {"unknown uid", []string{`type=USER_LOGIN msg=audit(1759572004.000:4280): pid=1 uid=0 auid=4242 ses=9 msg='op=login id=4242 exe="/usr/sbin/sshd" hostname=? addr=10.0.0.5 terminal=/dev/pts/1 res=success'`},
+ "srv", "USER_LOGIN", "info", "4242: login via /usr/sbin/sshd from 10.0.0.5 on /dev/pts/1 → success",
+ map[string]string{"user": "4242"}},
+ {"no result", []string{`type=CONFIG_CHANGE msg=audit(1759572005.000:4290): auid=1000 ses=3 op=add_rule key="watch" list=4`},
+ "srv", "CONFIG_CHANGE", "info", "cedric: config_change add_rule [watch]",
+ map[string]string{"res": ""}},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ ev := auditEv(t, tt.lines...)
+ e := auditEntry(ev, &userNames{path: passwd}, "srv", now)
+ if e.Host != tt.host || e.App != tt.app || e.Severity != tt.sev || e.SourceType != "audit" || !e.Wait {
+ t.Errorf("entry %+v", e)
+ }
+ if e.Message != tt.msg {
+ t.Errorf("message\n got %q\nwant %q", e.Message, tt.msg)
+ }
+ for k, v := range tt.fields {
+ if e.Extra[k] != v {
+ t.Errorf("%s = %q, want %q", k, e.Extra[k], v)
+ }
+ }
+ if e.Extra["audit_raw"] != strings.Join(tt.lines, "\n") {
+ t.Errorf("raw %q", e.Extra["audit_raw"])
+ }
+ if !e.Time.Equal(ev.Time) || e.Time.IsZero() {
+ t.Errorf("time %v", e.Time)
+ }
+ })
+ }
+}
+
+type entrySink struct {
+ mu sync.Mutex
+ es []*Entry
+}
+
+func (s *entrySink) add(e *Entry) {
+ s.mu.Lock()
+ s.es = append(s.es, e)
+ s.mu.Unlock()
+}
+
+func (s *entrySink) serials() []string {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ var out []string
+ for _, e := range s.es {
+ out = append(out, e.Extra["audit_serial"])
+ }
+ return out
+}
+
+func appendLines(t *testing.T, path string, lines ...string) {
+ t.Helper()
+ f, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer f.Close()
+ for _, l := range lines {
+ if _, err := f.WriteString(l + "\n"); err != nil {
+ t.Fatal(err)
+ }
+ }
+}
+
+func TestAuditFollow(t *testing.T) {
+ dir := t.TempDir()
+ logPath := filepath.Join(dir, "audit.log")
+ data := filepath.Join(dir, "data")
+ if err := os.Mkdir(data, 0o755); err != nil {
+ t.Fatal(err)
+ }
+ sink := &entrySink{}
+ newAuditT := func() *Audit {
+ a := NewAudit(logPath, dir, "srv", data, 100*365*24*time.Hour, sink.add)
+ a.cfg.Enabled = true
+ return a
+ }
+ a := newAuditT()
+ now := time.Unix(1759572100, 0)
+
+ // Missing file: reported, then read once it exists.
+ a.poll(now)
+ if st := a.Status(); st["code"] != "not_found" || st["open"] != false {
+ t.Fatalf("status %v", st)
+ }
+ appendLines(t, logPath, auditExec[:3]...)
+ appendLines(t, logPath, auditSudo)
+ a.poll(now)
+ if got := sink.serials(); len(got) != 0 { // events may still get records
+ t.Fatalf("after first poll: %v", got)
+ }
+ // A partial line is not read.
+ f, _ := os.OpenFile(logPath, os.O_APPEND|os.O_WRONLY, 0)
+ f.WriteString(auditSSH[:20])
+ f.Close()
+ a.poll(now.Add(5 * time.Second))
+ if got := sink.serials(); len(got) != 2 || got[0] != "4242" || got[1] != "4250" {
+ t.Fatalf("after second poll: %v", got)
+ }
+ if a.Status()["open"] != true || a.Status()["read"] != uint64(2) {
+ t.Fatalf("status %v", a.Status())
+ }
+ a.saveState()
+
+ // Restart: nothing is sent twice, the partial line is read once complete.
+ a = newAuditT()
+ f, _ = os.OpenFile(logPath, os.O_APPEND|os.O_WRONLY, 0)
+ f.WriteString(auditSSH[20:] + "\n")
+ f.Close()
+ a.poll(now.Add(10 * time.Second))
+ a.poll(now.Add(20 * time.Second))
+ if got := sink.serials(); len(got) != 3 || got[2] != "4260" {
+ t.Fatalf("after restart: %v", got)
+ }
+
+ // Rotation: lines added to the old file before the switch are not lost.
+ appendLines(t, logPath, auditOpen[0])
+ a.poll(now.Add(21 * time.Second)) // read, still open
+ appendLines(t, logPath, auditOpen[1])
+ if err := os.Rename(logPath, logPath+".1"); err != nil {
+ t.Fatal(err)
+ }
+ appendLines(t, logPath, strings.Replace(auditSudo, "4250", "4300", 1))
+ a.poll(now.Add(22 * time.Second))
+ a.poll(now.Add(30 * time.Second))
+ got := sink.serials()
+ if len(got) != 5 || got[3] != "4270" || got[4] != "4300" {
+ t.Fatalf("after rotation: %v", got)
+ }
+ sink.mu.Lock()
+ if n := len(sink.es[3].Extra["audit_raw"]); n == 0 || !strings.Contains(sink.es[3].Extra["audit_raw"], "/etc/shadow") {
+ t.Errorf("rotated event missing its PATH record: %q", sink.es[3].Extra["audit_raw"])
+ }
+ sink.mu.Unlock()
+}
+
+func TestAuditBackfillAndConfigure(t *testing.T) {
+ dir := t.TempDir()
+ logPath := filepath.Join(dir, "audit.log")
+ appendLines(t, logPath, auditExec...)
+ appendLines(t, logPath, auditSudo)
+ sink := &entrySink{}
+ // Only events newer than the backfill window are sent on the first read.
+ a := NewAudit(logPath, dir, "", dir, time.Hour, sink.add)
+ a.cfg.Enabled = true
+ a.poll(time.Unix(1759572001+7200, 0))
+ a.emit(a.asm.flushAll())
+ if got := sink.serials(); len(got) != 0 {
+ t.Fatalf("too old: %v", got)
+ }
+ a = NewAudit(logPath, dir, "", filepath.Join(dir, "other"), time.Hour, sink.add)
+ if err := os.Mkdir(filepath.Join(dir, "other"), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ a.cfg.Enabled = true
+ a.poll(time.Unix(1759572001+60, 0))
+ a.emit(a.asm.flushAll())
+ if got := sink.serials(); len(got) != 2 {
+ t.Fatalf("backfill: %v", got)
+ }
+ if e := sink.es[0]; e.Host != "localhost" {
+ t.Errorf("host %q", e.Host)
+ }
+
+ // Run picks the configuration up; turning it off resets the position.
+ ctx, cancel := context.WithCancel(context.Background())
+ done := make(chan struct{})
+ go func() { a.Run(ctx); close(done) }()
+ if err := a.Configure(auditConfig{Enabled: false}); err != nil {
+ t.Fatal(err)
+ }
+ deadline := time.Now().Add(2 * time.Second)
+ for time.Now().Before(deadline) {
+ if b, err := os.ReadFile(filepath.Join(dir, "other", "audit-state.json")); err == nil && strings.Contains(string(b), `"off": 0`) {
+ break
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+ cancel()
+ <-done
+ if st := a.Status(); st["enabled"] != false || st["defaultPath"] != logPath {
+ t.Fatalf("status %v", st)
+ }
+ b, _ := os.ReadFile(filepath.Join(dir, "other", "audit-state.json"))
+ if !strings.Contains(string(b), `"off": 0`) {
+ t.Fatalf("state %s", b)
+ }
+}
diff --git a/dmesg.go b/dmesg.go
index 63ec261..60cd303 100644
--- a/dmesg.go
+++ b/dmesg.go
@@ -7,7 +7,6 @@ import (
"errors"
"io"
"io/fs"
- "log"
"os"
"path/filepath"
"strconv"
@@ -173,7 +172,7 @@ func (d *Dmesg) saveState() {
pos := d.pos
d.mu.Unlock()
if err := writeJSONFile(d.statePath, pos); err != nil {
- log.Printf("dmesg: saving position: %v", err)
+ logEvent("dmesg", 4, "saving position: %v", err)
}
}
@@ -197,7 +196,7 @@ func (d *Dmesg) fail(err error) {
d.open, d.errCode, d.errDetail = false, code, detail
d.mu.Unlock()
if changed && code != "" {
- log.Printf("dmesg: %v", err)
+ logEvent("dmesg", 4, "cannot read the kernel messages: %v", err)
}
}
diff --git a/docker-compose.yml b/docker-compose.yml
index 857464a..ff372b2 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -38,6 +38,8 @@ services:
DOCKER_BACKFILL: ${DOCKER_BACKFILL:-1h}
HOST_LOGS_BACKFILL: ${HOST_LOGS_BACKFILL:-1h} # historique lu a l'activation des logs systeme de l'hote
DMESG_BACKFILL: ${DMESG_BACKFILL:-1h} # historique lu a l'activation des messages du noyau (dmesg)
+ AUDIT_BACKFILL: ${AUDIT_BACKFILL:-1h} # historique lu a l'activation de l'audit Linux (auditd)
+ AUDIT_HOST: ${AUDIT_HOST:-} # nom d'hote des evenements d'audit (vide : node= des enregistrements, sinon /host/etc/hostname)
group_add:
- "${HOST_LOGS_GID:-4}" # groupe autorise a lire les logs de l'hote (4 = adm sur Debian/Ubuntu)
# messages du noyau de l'hote (dmesg), source a activer dans Reglages > Sources :
@@ -55,6 +57,12 @@ services:
# logs systeme de l'hote, en lecture seule (source a activer dans Reglages > Sources)
- /var/log:/host/var/log:ro # journal systemd persistant et fichiers texte
- /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
+ # noms des utilisateurs et de l'hote pour les evenements d'audit (optionnel)
+ # - /etc/passwd:/host/etc/passwd:ro
+ # - /etc/hostname:/host/etc/hostname: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
diff --git a/docker.go b/docker.go
index 4aa49aa..e8ce1ff 100644
--- a/docker.go
+++ b/docker.go
@@ -242,7 +242,7 @@ func (m *DockerManager) setError(err error) {
m.mu.Lock()
defer m.mu.Unlock()
if m.lastErr != err.Error() {
- log.Printf("docker: %v", err)
+ logEvent("docker", 3, "Docker API unreachable: %v", err)
}
m.connected = false
m.lastErr = err.Error()
@@ -290,7 +290,7 @@ func (m *DockerManager) sync(ctx context.Context) {
m.mu.Lock()
defer m.mu.Unlock()
if !m.connected {
- log.Printf("docker: connected, %d containers", len(list))
+ logEvent("docker", 6, "connected to the Docker API, %d containers", len(list))
}
m.connected, m.lastErr = true, ""
seen := map[string]bool{}
@@ -371,7 +371,7 @@ func (m *DockerManager) follow(ctx context.Context, c DockerContainer) {
if ctx.Err() != nil {
return
}
- log.Printf("docker %s: %v", c.Name, err)
+ logEvent("docker", 4, "container %s: cannot inspect it: %v", c.Name, err)
sleepCtx(ctx, 5*time.Second)
continue
}
@@ -383,7 +383,7 @@ func (m *DockerManager) follow(ctx context.Context, c DockerContainer) {
return
}
if err != nil && !errors.Is(err, io.EOF) && !errors.Is(err, io.ErrUnexpectedEOF) {
- log.Printf("docker %s: %v", c.Name, err)
+ logEvent("docker", 4, "container %s: log stream lost, reconnecting: %v", c.Name, err)
}
sleepCtx(ctx, time.Second)
}
diff --git a/hostlogs.go b/hostlogs.go
index 2cca358..a9d64fe 100644
--- a/hostlogs.go
+++ b/hostlogs.go
@@ -7,7 +7,6 @@ import (
"errors"
"io"
"io/fs"
- "log"
"os"
"path/filepath"
"sort"
@@ -116,14 +115,18 @@ func (h *HostLogs) saveState() {
}
h.dirty = false
if err := writeJSONFile(h.statePath, h.pos); err != nil {
- log.Printf("host logs: saving positions: %v", err)
+ logEvent("host", 4, "saving positions: %v", err)
}
}
func (h *HostLogs) setStatus(mode string, files int, code, detail string) {
h.mu.Lock()
+ changed := code != h.errCode
h.mode, h.files, h.errCode, h.errDetail = mode, files, code, detail
h.mu.Unlock()
+ if changed && code != "" && code != "empty" {
+ logEvent("host", 4, "cannot read the host system logs (%s): %s", code, detail)
+ }
}
// scan reads what was added since the previous poll.
diff --git a/main.go b/main.go
index f87ad33..5a5003b 100644
--- a/main.go
+++ b/main.go
@@ -40,6 +40,9 @@ type config struct {
hostFill time.Duration
kmsgDevice string
kmsgFill time.Duration
+ auditLog string
+ auditHost string
+ auditFill time.Duration
batchSize int
queueSize int
flushEvery time.Duration
@@ -119,6 +122,9 @@ func main() {
hostFill: getenvDuration("HOST_LOGS_BACKFILL", time.Hour),
kmsgDevice: getenv("DMESG_DEVICE", "/dev/kmsg"),
kmsgFill: getenvDuration("DMESG_BACKFILL", time.Hour),
+ auditLog: getenv("AUDIT_LOG", "/var/log/audit/audit.log"),
+ auditHost: os.Getenv("AUDIT_HOST"),
+ auditFill: getenvDuration("AUDIT_BACKFILL", time.Hour),
batchSize: getenvInt("BATCH_SIZE", 1000),
queueSize: getenvInt("QUEUE_SIZE", 100000),
flushEvery: time.Duration(getenvInt("FLUSH_MS", 1000)) * time.Millisecond,
@@ -138,14 +144,16 @@ func main() {
var spool *Spool
if cfg.spoolMax > 0 {
if spool, err = OpenSpool(filepath.Join(cfg.dataDir, "spool"), cfg.spoolMax); err != nil {
- log.Printf("disk buffer disabled: %v", err)
+ log.Printf("disk buffer disabled: %v", err) // before the self-logs exist
spool = nil
}
}
store := NewStore(cfg.vlogsURL, cfg.batchSize, cfg.queueSize, cfg.flushEvery, spool)
+ // The store stops after everything else, so the stop event is still stored.
+ storeCtx, stopStore := context.WithCancel(context.Background())
storeDone := make(chan struct{})
go func() {
- store.Run(ctx)
+ store.Run(storeCtx)
close(storeDone)
}()
@@ -163,6 +171,20 @@ func main() {
hub.Publish(e)
})
}
+ // Internal events of LogStream (host=logstream), WARNING and above unless
+ // changed in Settings > Sources.
+ selfLog = NewSelfLog(cfg.dataDir, func(e *Entry) {
+ store.Enqueue(e)
+ hub.Publish(e)
+ })
+ selfCtx, stopSelf := context.WithCancel(context.Background())
+ selfDone := make(chan struct{})
+ go func() {
+ selfLog.Run(selfCtx)
+ close(selfDone)
+ }()
+ logEvent("main", 5, "LogStream started")
+
tcpMaxConns = getenvInt("SYSLOG_TCP_MAX_CONNS", tcpMaxConns)
if d := getenvDuration("SYSLOG_TCP_IDLE", tcpIdle); d > 0 {
tcpIdle = d
@@ -180,7 +202,7 @@ func main() {
if cfg.dockerLogs {
dm, err := NewDockerManager(cfg.dockerHost, cfg.dataDir, cfg.backfill, sink)
if err != nil {
- log.Printf("docker logs disabled: %v", err)
+ logEvent("docker", 3, "Docker logs disabled: %v", err)
} else {
go dm.Run(ctx)
api.docker = dm
@@ -193,6 +215,10 @@ func main() {
// Kernel messages of the host (/dev/kmsg), off until enabled in Settings > Sources.
api.dmesg = NewDmesg(cfg.kmsgDevice, cfg.dataDir, cfg.kmsgFill, sink)
go api.dmesg.Run(ctx)
+ // Linux audit log of the host (auditd), off until enabled in Settings > Sources.
+ api.audit = NewAudit(cfg.auditLog, cfg.hostRoot, cfg.auditHost, cfg.dataDir, cfg.auditFill, sink)
+ go api.audit.Run(ctx)
+ api.selfLog = selfLog
api.Routes(mux)
mux.Handle("GET /", http.FileServer(http.FS(static)))
@@ -229,6 +255,10 @@ func main() {
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
log.Fatalf("http: %v", err)
}
+ logEvent("main", 5, "LogStream stopping")
+ stopSelf()
+ <-selfDone
+ stopStore()
<-storeDone
log.Println("shutdown complete")
}
diff --git a/query.go b/query.go
index aea8147..bb9e866 100644
--- a/query.go
+++ b/query.go
@@ -119,10 +119,10 @@ func (f Filter) filterExpr() string {
parts = append(parts, "app:="+strconv.Quote(f.App))
}
switch f.Source {
- case "docker", "host", "kernel":
+ case "docker", "host", "kernel", "audit", "logstream":
parts = append(parts, `source_type:=`+strconv.Quote(f.Source))
case "syslog": // also matches logs stored before source_type existed
- parts = append(parts, `!(source_type:in("docker","host","kernel"))`)
+ parts = append(parts, `!(source_type:in("docker","host","kernel","audit","logstream"))`)
}
if f.Severity >= 0 && f.Severity < 7 {
names := make([]string, 0, 8)
diff --git a/selflog.go b/selflog.go
new file mode 100644
index 0000000..7f685a0
--- /dev/null
+++ b/selflog.go
@@ -0,0 +1,187 @@
+package main
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "log"
+ "os"
+ "path/filepath"
+ "sync"
+ "time"
+)
+
+// selfLogConfig is saved in /data/selflog.json and edited in Settings > Sources.
+type selfLogConfig struct {
+ Enabled bool `json:"enabled"`
+ MinLevel string `json:"min_level"` // severity name: "err", "warning", "notice", "info"…
+}
+
+// SelfLog turns the internal events of LogStream (storage errors, listener
+// failures, sources that stop…) into log entries, with host=logstream and
+// app=component. Events go through a bounded queue and a per-message
+// throttle: an event about a failing Store is itself stored through that
+// Store, so it must never be able to loop or pile up.
+type SelfLog struct {
+ cfgPath string
+ sink func(*Entry)
+ queue chan *Entry
+ interval time.Duration // one event per component and message format per interval
+
+ mu sync.Mutex
+ cfg selfLogConfig
+ last map[string]time.Time
+ suppressed map[string]int
+ emitted uint64
+ dropped uint64
+ throttled uint64
+}
+
+// selfLog receives the events of logEvent; nil (tests) only writes them to
+// the process log.
+var selfLog *SelfLog
+
+func NewSelfLog(dataDir string, sink func(*Entry)) *SelfLog {
+ s := &SelfLog{
+ cfgPath: filepath.Join(dataDir, "selflog.json"),
+ sink: sink,
+ queue: make(chan *Entry, 256),
+ interval: 30 * time.Second,
+ cfg: selfLogConfig{Enabled: true, MinLevel: "warning"},
+ last: map[string]time.Time{},
+ suppressed: map[string]int{},
+ }
+ if b, err := os.ReadFile(s.cfgPath); err == nil {
+ var cfg selfLogConfig
+ if json.Unmarshal(b, &cfg) == nil {
+ s.cfg = normSelfLogConfig(cfg)
+ }
+ }
+ return s
+}
+
+func normSelfLogConfig(cfg selfLogConfig) selfLogConfig {
+ if sevNum(cfg.MinLevel) < 0 {
+ cfg.MinLevel = "warning"
+ }
+ return cfg
+}
+
+func sevNum(name string) int {
+ for i, n := range severityNames {
+ if n == name {
+ return i
+ }
+ }
+ return -1
+}
+
+// logEvent writes an internal event to the process log and, when the
+// self-logs source is on and the level is high enough, emits it as an entry.
+// sev is a syslog severity (3 err, 4 warning, 5 notice, 6 info).
+func logEvent(component string, sev int, format string, args ...any) {
+ msg := fmt.Sprintf(format, args...)
+ log.Printf("%s: %s", component, msg)
+ if s := selfLog; s != nil {
+ s.Emit(component, sev, format, msg, time.Now())
+ }
+}
+
+// Emit queues an event without ever blocking. The same component and message
+// format are let through once per interval; the next one says how many were
+// left out in between.
+func (s *SelfLog) Emit(component string, sev int, format, msg string, now time.Time) {
+ sev = min(max(sev, 0), 7)
+ key := component + "\x00" + format
+ s.mu.Lock()
+ if !s.cfg.Enabled || sev > sevNum(s.cfg.MinLevel) {
+ s.mu.Unlock()
+ return
+ }
+ if t, ok := s.last[key]; ok && now.Sub(t) < s.interval {
+ s.suppressed[key]++
+ s.throttled++
+ s.mu.Unlock()
+ return
+ }
+ s.last[key] = now
+ if n := s.suppressed[key]; n > 0 {
+ msg += fmt.Sprintf(" (%d similar events not shown)", n)
+ delete(s.suppressed, key)
+ }
+ if len(s.last) > 1000 { // formats are constants: only a bug could get here
+ clear(s.last)
+ }
+ s.mu.Unlock()
+
+ e := &Entry{
+ Time: now,
+ Received: now,
+ Host: "logstream",
+ App: component,
+ Facility: "syslog",
+ Severity: severityNames[sev],
+ SevNum: sev,
+ Message: msg,
+ Proto: "internal",
+ SourceType: "logstream",
+ // Wait stays false: when the Store queue is full, the event is dropped
+ // instead of blocking the component that reports it.
+ }
+ select {
+ case s.queue <- e:
+ default:
+ s.mu.Lock()
+ s.dropped++
+ s.mu.Unlock()
+ }
+}
+
+// Run hands the queued events to the sink until the context ends, then
+// delivers what is left (the stop event among them).
+func (s *SelfLog) Run(ctx context.Context) {
+ for {
+ select {
+ case e := <-s.queue:
+ s.deliver(e)
+ case <-ctx.Done():
+ for {
+ select {
+ case e := <-s.queue:
+ s.deliver(e)
+ default:
+ return
+ }
+ }
+ }
+ }
+}
+
+func (s *SelfLog) deliver(e *Entry) {
+ s.sink(e)
+ s.mu.Lock()
+ s.emitted++
+ s.mu.Unlock()
+}
+
+// Configure saves and applies the configuration.
+func (s *SelfLog) Configure(cfg selfLogConfig) error {
+ cfg = normSelfLogConfig(cfg)
+ s.mu.Lock()
+ s.cfg = cfg
+ s.mu.Unlock()
+ return writeJSONFile(s.cfgPath, cfg)
+}
+
+// Status is the state shown in Settings > Sources.
+func (s *SelfLog) Status() map[string]any {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ return map[string]any{
+ "enabled": s.cfg.Enabled,
+ "min_level": s.cfg.MinLevel,
+ "emitted": s.emitted,
+ "dropped": s.dropped,
+ "throttled": s.throttled,
+ }
+}
diff --git a/selflog_test.go b/selflog_test.go
new file mode 100644
index 0000000..18bba4f
--- /dev/null
+++ b/selflog_test.go
@@ -0,0 +1,102 @@
+package main
+
+import (
+ "context"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+)
+
+func TestSelfLogEmit(t *testing.T) {
+ now := time.Unix(1759572000, 0)
+ tests := []struct {
+ name string
+ cfg selfLogConfig
+ events []int // severities, one second apart, same message format
+ want int // entries queued
+ suffix string
+ }{
+ {"warning and above", selfLogConfig{Enabled: true, MinLevel: "warning"}, []int{3}, 1, ""},
+ {"below the level", selfLogConfig{Enabled: true, MinLevel: "warning"}, []int{5, 6}, 0, ""},
+ {"info level", selfLogConfig{Enabled: true, MinLevel: "info"}, []int{6}, 1, ""},
+ {"disabled", selfLogConfig{Enabled: false, MinLevel: "debug"}, []int{0}, 0, ""},
+ {"bad level falls back to warning", normSelfLogConfig(selfLogConfig{Enabled: true, MinLevel: "loud"}), []int{4, 5}, 1, ""},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ s := NewSelfLog(t.TempDir(), func(*Entry) {})
+ s.cfg = tt.cfg
+ for i, sev := range tt.events {
+ s.Emit("store", sev, "fmt %d", "msg", now.Add(time.Duration(i)*time.Second))
+ }
+ if len(s.queue) != tt.want {
+ t.Fatalf("%d queued, want %d", len(s.queue), tt.want)
+ }
+ })
+ }
+
+ // Same component and format: once per interval, then a count of the others.
+ s := NewSelfLog(t.TempDir(), func(*Entry) {})
+ for i := 0; i < 5; i++ {
+ s.Emit("store", 3, "%d messages lost", "x", now.Add(time.Duration(i)*time.Second))
+ }
+ s.Emit("syslog", 3, "%d messages lost", "y", now) // another component
+ s.Emit("store", 3, "%d messages lost", "z", now.Add(s.interval))
+ if len(s.queue) != 3 {
+ t.Fatalf("%d queued", len(s.queue))
+ }
+ <-s.queue
+ if e := <-s.queue; e.App != "syslog" || e.Host != "logstream" || e.SourceType != "logstream" || e.Wait {
+ t.Fatalf("entry %+v", e)
+ }
+ if e := <-s.queue; e.Message != "z (4 similar events not shown)" || e.Severity != "err" {
+ t.Fatalf("message %q", e.Message)
+ }
+ if st := s.Status(); st["throttled"] != uint64(4) {
+ t.Fatalf("status %v", st)
+ }
+
+ // A full queue drops events instead of blocking.
+ s = NewSelfLog(t.TempDir(), func(*Entry) {})
+ for i := 0; i < cap(s.queue)+10; i++ {
+ s.Emit("store", 3, "event %d", "x", now.Add(time.Duration(i)*s.interval))
+ }
+ if st := s.Status(); st["dropped"] != uint64(10) {
+ t.Fatalf("status %v", st)
+ }
+}
+
+func TestSelfLogRunAndConfigure(t *testing.T) {
+ dir := t.TempDir()
+ sink := &entrySink{}
+ s := NewSelfLog(dir, sink.add)
+ if err := s.Configure(selfLogConfig{Enabled: true, MinLevel: "notice"}); err != nil {
+ t.Fatal(err)
+ }
+ if b, _ := os.ReadFile(filepath.Join(dir, "selflog.json")); !strings.Contains(string(b), `"notice"`) {
+ t.Fatalf("saved %s", b)
+ }
+ if s2 := NewSelfLog(dir, nil); s2.cfg.MinLevel != "notice" {
+ t.Fatalf("reloaded %+v", s2.cfg)
+ }
+ old := selfLog
+ selfLog = s
+ defer func() { selfLog = old }()
+
+ ctx, cancel := context.WithCancel(context.Background())
+ done := make(chan struct{})
+ go func() { s.Run(ctx); close(done) }()
+ logEvent("main", 5, "LogStream started")
+ logEvent("main", 6, "too verbose")
+ cancel()
+ <-done
+ logEvent("main", 5, "LogStream stopping") // queued after Run: not lost if Run drains it
+ sink.mu.Lock()
+ n := len(sink.es)
+ sink.mu.Unlock()
+ if n != 1 || s.Status()["emitted"] != uint64(1) {
+ t.Fatalf("%d entries, status %v", n, s.Status())
+ }
+}
diff --git a/store.go b/store.go
index 2598a37..e1a2c02 100644
--- a/store.go
+++ b/store.go
@@ -8,7 +8,6 @@ import (
"errors"
"fmt"
"io"
- "log"
"net/http"
"net/url"
"strings"
@@ -34,6 +33,7 @@ type Store struct {
ingested atomic.Int64
dropped atomic.Int64
lastErr atomic.Value // string
+ failing atomic.Bool // VictoriaLogs refused or missed the last send
}
func NewStore(base string, batchSize, queueSize int, flushEvery time.Duration, spool *Spool) *Store {
@@ -133,7 +133,7 @@ func (s *Store) store(ctx context.Context, batch []*Entry) {
if err != nil {
s.dropped.Add(int64(len(batch)))
s.lastErr.Store(err.Error())
- log.Printf("victorialogs: %d messages dropped: %v", len(batch), err)
+ logEvent("store", 3, "%d messages lost: %v", len(batch), err)
return
}
for _, e := range batch {
@@ -153,9 +153,10 @@ func (s *Store) save(ctx context.Context, body []byte, lines int) error {
}
if err = s.post(ctx, body); err == nil {
s.ingested.Add(int64(lines))
- s.lastErr.Store("")
+ s.sent()
return nil
}
+ s.sendFailed(err)
}
return err
}
@@ -164,10 +165,10 @@ func (s *Store) save(ctx context.Context, body []byte, lines int) error {
if waiting, _ := s.spool.Pending(); waiting == 0 {
if postErr = s.post(ctx, body); postErr == nil {
s.ingested.Add(int64(lines))
- s.lastErr.Store("")
+ s.sent()
return nil
}
- s.lastErr.Store(postErr.Error())
+ s.sendFailed(postErr)
}
err := s.spool.Write(body, lines)
if err == nil {
@@ -192,13 +193,13 @@ func (s *Store) save(ctx context.Context, body []byte, lines int) error {
// an increasing pause (up to 30 s) while it keeps failing.
func (s *Store) replay(ctx context.Context) {
if n, size := s.spool.Pending(); n > 0 {
- log.Printf("spool: %d messages (%d bytes) waiting from a previous run", n, size)
+ logEvent("store", 5, "disk spool: %d messages (%d bytes) waiting from a previous run", n, size)
}
backoff := time.Second
for {
f, body, ok, err := s.spool.Oldest()
if err != nil {
- log.Printf("spool: %v", err)
+ logEvent("store", 4, "disk spool: %v", err)
}
if !ok {
select {
@@ -214,17 +215,17 @@ func (s *Store) replay(ctx context.Context) {
case err == nil:
s.spool.Remove(f)
s.ingested.Add(f.lines)
- s.lastErr.Store("")
+ s.sent()
backoff = time.Second
continue
case isRejected(err):
// VictoriaLogs refuses the data itself: sending it again would not help.
s.spool.Remove(f)
s.dropped.Add(f.lines)
- log.Printf("spool: %d messages refused by victorialogs: %v", f.lines, err)
+ logEvent("store", 3, "disk spool: %d messages refused by VictoriaLogs and lost: %v", f.lines, err)
continue
}
- s.lastErr.Store(err.Error())
+ s.sendFailed(err)
if !sleepCtx(ctx, backoff) {
return
}
@@ -232,6 +233,26 @@ func (s *Store) replay(ctx context.Context) {
}
}
+// sendFailed and sent report when VictoriaLogs stops and starts taking the
+// batches again, once per change.
+func (s *Store) sendFailed(err error) {
+ s.lastErr.Store(err.Error())
+ if !s.failing.Swap(true) {
+ if s.spool != nil {
+ logEvent("store", 4, "VictoriaLogs unreachable, batches are kept in the disk spool: %v", err)
+ } else {
+ logEvent("store", 4, "VictoriaLogs unreachable, retrying: %v", err)
+ }
+ }
+}
+
+func (s *Store) sent() {
+ s.lastErr.Store("")
+ if s.failing.Swap(false) {
+ logEvent("store", 5, "VictoriaLogs reachable again")
+ }
+}
+
func encodeBatch(batch []*Entry) ([]byte, error) {
var buf bytes.Buffer
enc := json.NewEncoder(&buf)
diff --git a/syslog.go b/syslog.go
index 2145ebd..5d9ca62 100644
--- a/syslog.go
+++ b/syslog.go
@@ -36,7 +36,7 @@ type Entry struct {
Proto string // udp or tcp
HostIP string // original host value when it was an IP resolved through DNS
- SourceType string // "syslog", "docker", "host" or "kernel"
+ SourceType string // "syslog", "docker", "host", "kernel", "audit" or "logstream"
Extra map[string]string // additional fields (docker: container, image, …)
// Not stored. Wait: the producer can be slowed down when the queue is full
@@ -239,7 +239,7 @@ func serveUDP(ctx context.Context, pc net.PacketConn, sink func(*Entry)) {
if ctx.Err() != nil {
return
}
- log.Printf("syslog udp: %v", err)
+ logEvent("syslog", 4, "UDP read error: %v", err)
time.Sleep(100 * time.Millisecond)
continue
}
@@ -262,14 +262,14 @@ func serveTCP(ctx context.Context, ln net.Listener, sink func(*Entry)) {
if ctx.Err() != nil {
return
}
- log.Printf("syslog tcp: %v", err)
+ logEvent("syslog", 4, "TCP accept error: %v", err)
time.Sleep(100 * time.Millisecond)
continue
}
select {
case slots <- struct{}{}:
default:
- log.Printf("syslog tcp: %d connections already open, refusing %s", tcpMaxConns, conn.RemoteAddr())
+ logEvent("syslog", 4, "%d TCP connections already open, refusing %s", tcpMaxConns, conn.RemoteAddr())
conn.Close()
continue
}
diff --git a/syslogserver.go b/syslogserver.go
index 2703acb..f085d1a 100644
--- a/syslogserver.go
+++ b/syslogserver.go
@@ -94,7 +94,7 @@ func (s *SyslogServer) apply() {
pc, err := net.ListenPacket("udp", s.addr)
if err != nil {
s.errUDP = err.Error()
- log.Printf("syslog udp: %v", err)
+ logEvent("syslog", 3, "UDP listener failed: %v", err)
} else {
if uc, ok := pc.(*net.UDPConn); ok {
_ = uc.SetReadBuffer(8 << 20)
@@ -108,7 +108,7 @@ func (s *SyslogServer) apply() {
ln, err := net.Listen("tcp", s.addr)
if err != nil {
s.errTCP = err.Error()
- log.Printf("syslog tcp: %v", err)
+ logEvent("syslog", 3, "TCP listener failed: %v", err)
} else {
s.ln = ln
go serveTCP(ctx, ln, s.sink)
diff --git a/web/app.js b/web/app.js
index ec61169..319af1f 100644
--- a/web/app.js
+++ b/web/app.js
@@ -78,6 +78,22 @@ const I18N = {
sourceAria: 'Source', srcAll: 'All sources', tabSources: 'Sources',
srcHost: 'Host system',
srcKernel: 'Kernel (dmesg)',
+ auditTitle: 'Linux audit (auditd)',
+ auditEnabled: 'Collect the audit events of this machine',
+ auditPath: 'File',
+ auditOff: 'Off: the audit log of the machine hosting LogStream is not read.',
+ auditOk: ({ p, r }) => `Following ${p}: ${r} events since startup.`,
+ audit_not_found: 'Audit log not found: mount /var/log/audit in the container (see docker-compose.yml) and check the path. ',
+ audit_permission: 'Permission denied on the audit log: set log_group in /etc/audit/auditd.conf, or add an ACL for UID 10001 (see the README). ',
+ audit_read: 'Cannot read the audit log: ',
+ auditHelp: 'Reads the log of the Linux audit daemon (auditd): logins, sudo, watched files, syscalls matched by your rules… The records of one event are grouped in one line, with a summary (who, what, result) and the raw record below it; failures are warnings. When turned on, the last hour is read first (AUDIT_BACKFILL); rotations are followed and reading resumes where it stopped.',
+ selfTitle: 'LogStream internal logs',
+ selfEnabled: 'Record the events of LogStream itself',
+ selfLevel: 'Minimum level',
+ selfOff: 'Off: the events of LogStream only go to its container output.',
+ selfOk: ({ n }) => `${n} events recorded since startup (host logstream).`,
+ selfDropped: (n) => ` ${n} left out (repeated or too many at once).`,
+ selfHelp: 'Turns the events of LogStream into logs (host logstream, app = component): VictoriaLogs unreachable and back, lost batches, syslog listener errors, Docker streams lost, failing sources, start and stop. The same event is recorded at most once every 30 s, with the number of repeats.',
hostTitle: 'Host system logs',
hostEnabled: 'Collect the system logs of this machine',
hostOff: 'Off: the system logs of the machine hosting LogStream are not collected.',
@@ -262,6 +278,22 @@ const I18N = {
sourceAria: 'Source', srcAll: 'Toutes les sources', tabSources: 'Sources',
srcHost: 'Système hôte',
srcKernel: 'Noyau (dmesg)',
+ auditTitle: 'Audit Linux (auditd)',
+ auditEnabled: 'Collecter les événements d\'audit de cette machine',
+ auditPath: 'Fichier',
+ auditOff: 'Désactivé : le journal d\'audit de la machine qui héberge LogStream n\'est pas lu.',
+ auditOk: ({ p, r }) => `Suivi de ${p} : ${r} événements depuis le démarrage.`,
+ audit_not_found: 'Journal d\'audit introuvable : montez /var/log/audit dans le conteneur (voir docker-compose.yml) et vérifiez le chemin. ',
+ audit_permission: 'Accès refusé au journal d\'audit : réglez log_group dans /etc/audit/auditd.conf, ou ajoutez une ACL pour l\'UID 10001 (voir le README). ',
+ audit_read: 'Lecture du journal d\'audit impossible : ',
+ auditHelp: 'Lit le journal du démon d\'audit Linux (auditd) : connexions, sudo, fichiers surveillés, appels système visés par vos règles… Les enregistrements d\'un même événement sont regroupés sur une ligne, avec un résumé (qui, quoi, résultat) et l\'enregistrement brut en dessous ; les échecs sont des warnings. À l\'activation, la dernière heure est lue d\'abord (AUDIT_BACKFILL) ; les rotations sont suivies et la lecture reprend où elle s\'était arrêtée.',
+ selfTitle: 'Logs internes de LogStream',
+ selfEnabled: 'Enregistrer les événements de LogStream lui-même',
+ selfLevel: 'Niveau minimal',
+ selfOff: 'Désactivé : les événements de LogStream ne vont que dans la sortie de son conteneur.',
+ selfOk: ({ n }) => `${n} événements enregistrés depuis le démarrage (hôte logstream).`,
+ selfDropped: (n) => ` ${n} écartés (répétés ou trop nombreux d\'un coup).`,
+ selfHelp: 'Transforme les événements de LogStream en logs (hôte logstream, app = composant) : VictoriaLogs injoignable puis de retour, lots perdus, erreurs du listener syslog, flux Docker décrochés, sources en échec, démarrage et arrêt. Un même événement est enregistré au plus une fois toutes les 30 s, avec le nombre de répétitions.',
hostTitle: 'Logs système de l\'hôte',
hostEnabled: 'Collecter les logs système de cette machine',
hostOff: 'Désactivé : les logs système de la machine qui héberge LogStream ne sont pas collectés.',
@@ -564,6 +596,8 @@ function setLang(next) {
renderSyslog();
renderHost();
renderDmesg();
+ renderAudit();
+ renderSelfLog();
renderDocker();
}
@@ -871,10 +905,15 @@ function fmtTime(s) {
return `${day} ${p.h}:${p.m}:${p.s}.${p.ms}`;
}
+// Sources with their own look in the list: a colored edge and a badge
+// before the app name (the severity badge stays the same).
+const SRC_BADGE = { audit: 'AUDIT', logstream: 'LS' };
+
function rowHTML(r, isNew) {
const sev = r.severity || '';
const cls = SEV_CLASS[sev] || 'info';
- return ``
+ const badge = SRC_BADGE[r.source_type];
+ return ``
+ ``
+ ``
+ `${esc(sev || '—')}`
@@ -882,13 +921,20 @@ function rowHTML(r, isNew) {
? `${esc(r.host_name || r.host)}`
: '')
+ (r.app
- ? `${r.source_type === 'docker' ? DOCKER_ICON : ''}${esc(r.app)}`
+ ? `${r.source_type === 'docker' ? DOCKER_ICON : ''}${badge ? `${badge}` : ''}${esc(r.app)}`
: '')
+ `${codesHTML(r._msg)}`
- + `
${highlight(r._msg)}
`
+ + `
${msgHTML(r)}
`
+ '';
}
+// Message cell; audit events also show their raw record (hidden in compact
+// density until the row is opened).
+function msgHTML(r) {
+ return highlight(r._msg)
+ + (r.source_type === 'audit' && r.audit_raw ? `