guard/internal/source/source.go

// Package source reads the lines Guard watches: log files followed like
// tail -F, the systemd journal (through journalctl), and Housecarl's log
// of dropped packets.
package source

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/netip"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	"guard/internal/housecarl"
)

// Line is one line of a log, with its time: when it was written (the
// journal, Housecarl's log) or read (a file).
type Line struct {
	Text string
	Time time.Time
	// Addr is the address, for sources that know it without a pattern
	// (Housecarl's log).
	Addr netip.Addr
}

// Source sends its lines until ctx ends.
type Source interface {
	Run(ctx context.Context, out chan<- Line) error
}

// maxLine is the longest line kept; longer ones are cut.
const maxLine = 64 << 10

// ---- files

// File follows a file, or every file a glob matches, from its end: lines
// written from now on. A file that appears later is read from its start;
// one that is rotated (replaced) is finished, then the new one is read;
// one that is truncated is read again from its start.
type File struct {
	Pattern string
	// Poll is how often the files are checked (default 1s).
	Poll time.Duration
	Log  *slog.Logger
	// Now gives the lines' time (tests).
	Now func() time.Time
}

type tailer struct {
	path    string
	f       *os.File
	info    os.FileInfo
	partial []byte
}

// Run follows the files.
func (s *File) Run(ctx context.Context, out chan<- Line) error {
	poll := s.Poll
	if poll == 0 {
		poll = time.Second
	}
	now := s.Now
	if now == nil {
		now = time.Now
	}
	tails := map[string]*tailer{}
	defer func() {
		for _, t := range tails {
			if t.f != nil {
				t.f.Close()
			}
		}
	}()
	first, warned := true, false
	for {
		paths, _ := filepath.Glob(s.Pattern)
		if len(paths) == 0 && !warned && s.Log != nil {
			s.Log.Warn("no file to watch yet; checking every few seconds", "file", s.Pattern)
			warned = true
		}
		for _, p := range paths {
			if _, ok := tails[p]; !ok {
				t := &tailer{path: p}
				// Files there when Guard starts are read from their end,
				// files that appear later from their start.
				if err := t.open(first); err != nil {
					if s.Log != nil {
						s.Log.Warn("can't read the file", "file", p, "err", err)
					}
					continue
				}
				tails[p] = t
				warned = false
			}
		}
		first = false
		for p, t := range tails {
			if err := t.step(ctx, out, now); err != nil {
				if errors.Is(err, context.Canceled) {
					return nil
				}
				if s.Log != nil {
					s.Log.Warn("reading the file failed", "file", p, "err", err)
				}
				t.f.Close()
				delete(tails, p)
			}
		}
		select {
		case <-ctx.Done():
			return nil
		case <-time.After(poll):
		}
	}
}

func (t *tailer) open(atEnd bool) error {
	f, err := os.Open(t.path)
	if err != nil {
		return err
	}
	info, err := f.Stat()
	if err != nil {
		f.Close()
		return err
	}
	if atEnd {
		if _, err := f.Seek(0, io.SeekEnd); err != nil {
			f.Close()
			return err
		}
	}
	t.f, t.info, t.partial = f, info, nil
	return nil
}

// step sends what was written since the last step, and follows a
// rotation or truncation.
func (t *tailer) step(ctx context.Context, out chan<- Line, now func() time.Time) error {
	if err := t.drain(ctx, out, now); err != nil {
		return err
	}
	info, err := os.Stat(t.path)
	if err != nil {
		// Rotated away, and the new file isn't there yet: keep the old.
		return nil
	}
	if !os.SameFile(info, t.info) {
		t.f.Close()
		return t.open(false)
	}
	if pos, err := t.f.Seek(0, io.SeekCurrent); err == nil && info.Size() < pos {
		// Truncated (copytruncate): from the start again.
		if _, err := t.f.Seek(0, io.SeekStart); err != nil {
			return err
		}
		t.partial = nil
		t.info = info
	}
	return nil
}

func (t *tailer) drain(ctx context.Context, out chan<- Line, now func() time.Time) error {
	buf := make([]byte, 32<<10)
	for {
		n, err := t.f.Read(buf)
		if n > 0 {
			data := append(t.partial, buf[:n]...)
			for {
				i := bytes.IndexByte(data, '\n')
				if i < 0 {
					break
				}
				line := data[:i]
				data = data[i+1:]
				if len(line) > maxLine {
					line = line[:maxLine]
				}
				select {
				case out <- Line{Text: strings.TrimRight(string(line), "\r"), Time: now()}:
				case <-ctx.Done():
					return ctx.Err()
				}
			}
			if len(data) > maxLine {
				data = data[:maxLine]
			}
			t.partial = append([]byte(nil), data...)
		}
		if err == io.EOF || n == 0 {
			return nil
		}
		if err != nil {
			return err
		}
	}
}

// ---- the journal

// Journal follows the systemd journal's entries of some units and/or
// syslog identifiers, through journalctl (no libsystemd: Guard has no
// cgo). It starts with what is written from now on; when journalctl ends
// it is started again after the last entry seen.
type Journal struct {
	Units, Identifiers []string
	// Command is journalctl (GUARD_JOURNALCTL overrides it, for tests).
	Command string
	Log     *slog.Logger
}

// JournalCommand is the journalctl to run.
func JournalCommand() string {
	if c := os.Getenv("GUARD_JOURNALCTL"); c != "" {
		return c
	}
	return "journalctl"
}

// JournalArgs are journalctl's arguments: following from now, or after a
// cursor; or (follow false) the last n entries.
func JournalArgs(units, idents []string, follow bool, cursor string, n int) []string {
	args := []string{"--output=json", "--no-pager"}
	switch {
	case !follow:
		args = append(args, "--lines="+strconv.Itoa(n))
	case cursor != "":
		args = append(args, "--follow", "--lines=all", "--after-cursor="+cursor)
	default:
		args = append(args, "--follow", "--lines=0")
	}
	for _, u := range units {
		args = append(args, "--unit="+u)
	}
	for _, i := range idents {
		args = append(args, "--identifier="+i)
	}
	return args
}

// Run follows the journal.
func (s *Journal) Run(ctx context.Context, out chan<- Line) error {
	cmd := s.Command
	if cmd == "" {
		cmd = JournalCommand()
	}
	cursor := ""
	for {
		c := exec.CommandContext(ctx, cmd, JournalArgs(s.Units, s.Identifiers, true, cursor, 0)...)
		var stderr bytes.Buffer
		c.Stderr = &stderr
		stdout, err := c.StdoutPipe()
		if err != nil {
			return err
		}
		if err := c.Start(); err != nil {
			return fmt.Errorf("starting %s: %w", cmd, err)
		}
		cursor = ReadJournal(ctx, stdout, out, cursor)
		err = c.Wait()
		if ctx.Err() != nil {
			return nil
		}
		if s.Log != nil {
			s.Log.Warn("journalctl ended; starting it again in 5 seconds", "err", err, "stderr", strings.TrimSpace(stderr.String()))
		}
		select {
		case <-ctx.Done():
			return nil
		case <-time.After(5 * time.Second):
		}
	}
}

// ReadJournal sends the messages of journalctl's JSON output, and returns
// the last entry's cursor (or cursor, if there was none).
func ReadJournal(ctx context.Context, r io.Reader, out chan<- Line, cursor string) string {
	sc := bufio.NewScanner(r)
	sc.Buffer(make([]byte, 64<<10), 4<<20)
	for sc.Scan() {
		l, c, ok := ParseJournal(sc.Bytes())
		if c != "" {
			cursor = c
		}
		if !ok {
			continue
		}
		select {
		case out <- l:
		case <-ctx.Done():
			return cursor
		}
	}
	return cursor
}

// ParseJournal reads one entry of journalctl --output=json: its message
// (a string, or bytes when it isn't UTF-8), time and cursor.
func ParseJournal(data []byte) (Line, string, bool) {
	var e struct {
		Message  json.RawMessage `json:"MESSAGE"`
		Cursor   string          `json:"__CURSOR"`
		Realtime string          `json:"__REALTIME_TIMESTAMP"`
	}
	if json.Unmarshal(data, &e) != nil {
		return Line{}, "", false
	}
	var text string
	if json.Unmarshal(e.Message, &text) != nil {
		var b []byte
		var ints []int
		if json.Unmarshal(e.Message, &ints) != nil {
			return Line{}, e.Cursor, false
		}
		for _, n := range ints {
			b = append(b, byte(n))
		}
		text = string(b)
	}
	if len(text) > maxLine {
		text = text[:maxLine]
	}
	t := time.Now()
	if us, err := strconv.ParseInt(e.Realtime, 10, 64); err == nil {
		t = time.UnixMicro(us)
	}
	return Line{Text: text, Time: t}, e.Cursor, true
}

// ---- Housecarl's log

// LogReader reads Housecarl's log of packets.
type LogReader interface {
	Log(ctx context.Context, after time.Time) ([]housecarl.LogEntry, error)
}

// HousecarlDrops follows the packets Housecarl's rules dropped (or
// rejected) and logged, from now on. Protocol limits them ("tcp",
// "udp", "any"); Shadow counts shadow mode's, which weren't dropped.
type HousecarlDrops struct {
	Client   LogReader
	Protocol string
	Shadow   bool
	Poll     time.Duration
	Log      *slog.Logger
}

// Run follows the log.
func (s *HousecarlDrops) Run(ctx context.Context, out chan<- Line) error {
	poll := s.Poll
	if poll == 0 {
		poll = 2 * time.Second
	}
	after := time.Now()
	failing := false
	for {
		entries, err := s.Client.Log(ctx, after)
		switch {
		case err != nil && !failing:
			if s.Log != nil {
				s.Log.Warn("reading Housecarl's log failed (the token needs log:read); trying again", "err", err)
			}
			failing = true
		case err == nil:
			failing = false
			// Newest first: send them oldest first.
			for i := len(entries) - 1; i >= 0; i-- {
				e := entries[i]
				if e.Time.After(after) {
					after = e.Time
				}
				if l, ok := s.line(e); ok {
					select {
					case out <- l:
					case <-ctx.Done():
						return nil
					}
				}
			}
		}
		select {
		case <-ctx.Done():
			return nil
		case <-time.After(poll):
		}
	}
}

func (s *HousecarlDrops) line(e housecarl.LogEntry) (Line, bool) {
	if e.Verdict != "drop" && e.Verdict != "reject" || e.Shadow && !s.Shadow {
		return Line{}, false
	}
	if p := strings.ToLower(s.Protocol); p != "any" && p != "" && e.Protocol != p {
		return Line{}, false
	}
	a, err := netip.ParseAddr(e.Source)
	if err != nil {
		return Line{}, false
	}
	text := fmt.Sprintf("%s %s %s -> %s", strings.TrimSpace(strings.TrimSuffix(strings.TrimSpace(e.Prefix), ":")), e.Protocol,
		hostPort(e.Source, e.SrcPort), hostPort(e.Dest, e.DestPort))
	return Line{Text: text, Time: e.Time, Addr: a.Unmap()}, true
}

func hostPort(a string, p int) string {
	if p == 0 {
		return a
	}
	if strings.Contains(a, ":") {
		return "[" + a + "]:" + strconv.Itoa(p)
	}
	return a + ":" + strconv.Itoa(p)
}