feat: DTail fork — server/client feature development

Squashed development of the snonux/dtail fork's product code (internal/, cmd/)
since diverging from mimecast/dtail. Major areas:

- Read/output path: the former "turbo" channel-less path is now the single,
  default server-side read/output path for cat/grep/tail and MapReduce; the old
  channel-based path and its config/env toggles were removed.
- MapReduce: single aggregate implementation (server + serverless) fed directly
  by a processor pipeline, with input-exhausted finalization via the shutdown
  coordinator; high-concurrency and data-race fixes.
- Journal source reads (journal:unit.service) via journalctl, Linux-gated behind
  a journal-v1 capability.
- Auth-key fast reconnect: in-memory per-user public-key cache with TTL/max-keys,
  registered over an authenticated session (AUTHKEY), checked before
  authorized_keys.
- Interactive query reload (--interactive-query) with SESSION START/UPDATE
  generation boundaries and capability negotiation.
- Client-side deadlines: --timeout / --shutdownAfter as context deadlines;
  follow shutdown handling.
- Client logging: diagnostics-only daily log by default, opt-in payload tee via
  --log-payload.
- Numerous correctness fixes (buffer-pool double-recycle races, EOF-sentinel
  leaks, glob-expansion cap, TOCTOU in CSV parsing) with accompanying unit tests.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

2 files changed, 288 insertions, 37 deletions

diff --git a/internal/user/server/user.go b/internal/user/server/user.go
index d391672..5925636 100644
--- a/internal/user/server/user.go
+++ b/internal/user/server/user.go
@@ -2,18 +2,16 @@ package server
 
 import (
 	"fmt"
-	"os"
 	"path/filepath"
 	"regexp"
 	"strings"
 
 	"github.com/mimecast/dtail/internal/config"
 	"github.com/mimecast/dtail/internal/io/dlog"
+	"github.com/mimecast/dtail/internal/io/fs"
 	"github.com/mimecast/dtail/internal/io/fs/permissions"
 )
 
-const maxLinkDepth int = 100
-
 // User represents an end-user which connected to the server via the DTail client.
 type User struct {
 	// The user name.
@@ -24,11 +22,20 @@ type User struct {
 	permissions []string
 }
 
+// PermissionLookup resolves permissions for a given SSH user.
+type PermissionLookup func(string) ([]string, error)
+
 // New returns a new user.
-func New(name, remoteAddress string) (*User, error) {
-	permissions, err := config.ServerUserPermissions(name)
-	if err != nil {
-		return nil, err
+func New(name, remoteAddress string, permissionLookup PermissionLookup) (*User, error) {
+	var (
+		permissions []string
+		err         error
+	)
+	if permissionLookup != nil {
+		permissions, err = permissionLookup(name)
+		if err != nil {
+			return nil, err
+		}
 	}
 	return &User{
 		Name:          name,
@@ -43,27 +50,30 @@ func (u *User) String() string {
 }
 
 // HasFilePermission is used to determine whether user is allowed to read a file.
-func (u *User) HasFilePermission(filePath, permissionType string) (hasPermission bool) {
+func (u *User) HasFilePermission(filePath, permissionType string) bool {
+	_, hasPermission := u.ValidateReadTarget(filePath, permissionType)
+	return hasPermission
+}
+
+// ValidateReadTarget resolves and authorizes a file path for server-side reads.
+func (u *User) ValidateReadTarget(filePath, permissionType string) (fs.ValidatedReadTarget, bool) {
 	dlog.Server.Debug(u, filePath, permissionType, "Checking config permissions")
-	if u.Name == config.ScheduleUser || u.Name == config.ContinuousUser {
-		// Background user has same permissions as dtail process itself.
-		return true
+	if fs.IsJournalSpec(filePath) {
+		return u.validateJournalReadTarget(filePath, permissionType)
 	}
 
 	cleanPath, err := filepath.EvalSymlinks(filePath)
 	if err != nil {
 		dlog.Server.Error(u, filePath, permissionType,
 			"Unable to evaluate symlinks", err)
-		hasPermission = false
-		return
+		return fs.ValidatedReadTarget{}, false
 	}
 
 	cleanPath, err = filepath.Abs(cleanPath)
 	if err != nil {
 		dlog.Server.Error(u, cleanPath, permissionType,
 			"Unable to make file path absolute", err)
-		hasPermission = false
-		return
+		return fs.ValidatedReadTarget{}, false
 	}
 
 	if cleanPath != filePath {
@@ -71,11 +81,42 @@ func (u *User) HasFilePermission(filePath, permissionType string) (hasPermission
 			"Calculated new clean path from original file path (possibly symlink)")
 	}
 
-	hasPermission, err = u.hasFilePermission(cleanPath, permissionType)
+	if u.Name != config.ScheduleUser && u.Name != config.ContinuousUser {
+		hasPermission, permissionErr := u.hasFilePermission(cleanPath, permissionType)
+		if permissionErr != nil {
+			dlog.Server.Warn(u, cleanPath, permissionErr)
+		}
+		if !hasPermission {
+			return fs.ValidatedReadTarget{}, false
+		}
+	}
+
+	target, err := fs.NewValidatedReadTarget(cleanPath)
 	if err != nil {
-		dlog.Server.Warn(u, cleanPath, err)
+		dlog.Server.Warn(u, cleanPath, permissionType, "Unable to validate read target", err)
+		return fs.ValidatedReadTarget{}, false
 	}
-	return
+
+	return target, true
+}
+
+func (u *User) validateJournalReadTarget(spec, permissionType string) (fs.ValidatedReadTarget, bool) {
+	if u.Name != config.ScheduleUser && u.Name != config.ContinuousUser {
+		hasPermission, permissionErr := u.iteratePaths(spec, permissionType)
+		if permissionErr != nil {
+			dlog.Server.Warn(u, spec, permissionErr)
+		}
+		if !hasPermission {
+			return fs.ValidatedReadTarget{}, false
+		}
+	}
+
+	target, err := fs.NewValidatedJournalTarget(spec)
+	if err != nil {
+		dlog.Server.Warn(u, spec, permissionType, "Unable to validate journal read target", err)
+		return fs.ValidatedReadTarget{}, false
+	}
+	return target, true
 }
 
 func (u *User) hasFilePermission(cleanPath, permissionType string) (bool, error) {
@@ -86,14 +127,6 @@ func (u *User) hasFilePermission(cleanPath, permissionType string) (bool, error)
 	dlog.Server.Info(u, cleanPath, permissionType,
 		"User with OS file system permissions to path")
 
-	// Only allow to follow regular files or symlinks.
-	info, err := os.Lstat(cleanPath)
-	if err != nil {
-		return false, fmt.Errorf("Unable to determine file type: %w", err)
-	}
-	if !info.Mode().IsRegular() {
-		return false, fmt.Errorf("Can only open regular files or follow symlinks")
-	}
 	hasPermission, err := u.iteratePaths(cleanPath, permissionType)
 	if err != nil {
 		return false, err
@@ -102,14 +135,24 @@ func (u *User) hasFilePermission(cleanPath, permissionType string) (bool, error)
 	return hasPermission, nil
 }
 
+// iteratePaths evaluates the user's permission list against cleanPath for the
+// given permissionType and returns whether access is granted.
+//
+// Semantics — "deny wins":
+//   - The list is scanned in order.  A rule prefixed with '!' is a deny rule;
+//     any other rule is an allow rule.
+//   - As soon as a deny rule matches, the function returns false immediately.
+//     No later allow rule can override a deny — this prevents misconfigured
+//     ACL lists from accidentally granting access to sensitive paths.
+//   - If no deny rule matches but at least one allow rule does, access is granted.
+//   - If no rule matches at all, access is denied (deny-by-default).
 func (u *User) iteratePaths(cleanPath, permissionType string) (bool, error) {
-	// By default assume no permissions
+	// Default: no permission until a matching allow rule is found.
 	hasPermission := false
-	for _, permission := range u.permissions {
-		typeStr := "readfiles" // Assume ReadFiles by default.
-		var regexStr string
-		var negate bool
 
+	for _, permission := range u.permissions {
+		// Determine the permission type prefix; default is "readfiles".
+		typeStr := "readfiles"
 		splitted := strings.Split(permission, ":")
 		if len(splitted) > 1 {
 			typeStr = splitted[0]
@@ -121,10 +164,12 @@ func (u *User) iteratePaths(cleanPath, permissionType string) (bool, error) {
 			continue
 		}
 
-		regexStr = permission
-		if strings.HasPrefix(permission, "!") {
+		// Detect deny rules (prefixed with '!') and strip the prefix before
+		// compiling the regex.
+		negate := strings.HasPrefix(permission, "!")
+		regexStr := permission
+		if negate {
 			regexStr = permission[1:]
-			negate = true
 		}
 
 		re, err := regexp.Compile(regexStr)
@@ -132,11 +177,14 @@ func (u *User) iteratePaths(cleanPath, permissionType string) (bool, error) {
 			return false, fmt.Errorf("Permission test failed, can't compile regex "+
 				"'%s': %w", regexStr, err)
 		}
+
 		if negate && re.MatchString(cleanPath) {
-			dlog.Server.Info(u, cleanPath, "Permission test failed partially, "+
-				"matching negative pattern '%s'", permission)
-			hasPermission = false
+			// Deny rule matched: return false immediately (deny wins).
+			// A subsequent allow rule must never override an explicit deny.
+			dlog.Server.Info(u, cleanPath, "Permission denied: matching deny pattern", permission)
+			return false, nil
 		}
+
 		if !negate && re.MatchString(cleanPath) {
 			dlog.Server.Info(u, cleanPath, "Permission test passed partially, "+
 				"matching positive pattern", permission)
diff --git a/internal/user/server/user_test.go b/internal/user/server/user_test.go
new file mode 100644
index 0000000..a431d37
--- /dev/null
+++ b/internal/user/server/user_test.go
@@ -0,0 +1,203 @@
+package server
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	"github.com/mimecast/dtail/internal/config"
+	"github.com/mimecast/dtail/internal/io/dlog"
+	"github.com/mimecast/dtail/internal/io/fs"
+	"github.com/mimecast/dtail/internal/source"
+)
+
+// ensureTestDeps initialises config and the dlog server logger once per test
+// binary run. It is safe to call from multiple parallel tests because it
+// guards initialisation with the nil-checks used elsewhere in the test suite
+// (see internal/mapr/server/aggregate_test.go).
+func ensureTestDeps(t *testing.T) {
+	t.Helper()
+	if config.Server == nil {
+		config.Server = &config.ServerConfig{}
+	}
+	if config.Common == nil {
+		config.Common = &config.CommonConfig{
+			// Use "none" logger to suppress output during tests and avoid
+			// the factory panic caused by an empty logger name.
+			Logger:   "none",
+			LogLevel: "error",
+		}
+	}
+	if dlog.Server == nil {
+		ctx, cancel := context.WithCancel(context.Background())
+		t.Cleanup(cancel)
+		var wg sync.WaitGroup
+		wg.Add(1)
+		dlog.Start(ctx, &wg, source.Server)
+	}
+}
+
+// newTestUser creates a User with the given permission strings for testing.
+// It bypasses PermissionLookup and OS-level checks so the unit tests
+// focus solely on iteratePaths logic.
+func newTestUser(perms []string) *User {
+	return &User{
+		Name:          "testuser",
+		remoteAddress: "127.0.0.1",
+		permissions:   perms,
+	}
+}
+
+// TestIteratePaths_DenyWins verifies that a deny rule (prefix '!') wins over
+// any subsequent allow rule for the same path. This is the canonical
+// "deny-wins" / "first-matching-deny" semantics expected from an ACL system.
+//
+// Before the fix, the loop used last-match-wins semantics, so a trailing
+// allow rule could silently neutralise an earlier deny — a security footgun.
+func TestIteratePaths_DenyWins(t *testing.T) {
+	ensureTestDeps(t)
+	t.Parallel()
+
+	tests := []struct {
+		name        string
+		permissions []string
+		path        string
+		permType    string
+		wantPerm    bool
+	}{
+		{
+			// Deny rule comes first; a later allow rule must NOT override it.
+			name: "deny_then_allow_same_path_deny_wins",
+			permissions: []string{
+				"readfiles:!/var/log/secret.*",
+				"readfiles:/var/log/.*",
+			},
+			path:     "/var/log/secret.log",
+			permType: "readfiles",
+			wantPerm: false,
+		},
+		{
+			// Allow rule comes first, then a deny rule — deny must still win.
+			name: "allow_then_deny_same_path_deny_wins",
+			permissions: []string{
+				"readfiles:/var/log/.*",
+				"readfiles:!/var/log/secret.*",
+			},
+			path:     "/var/log/secret.log",
+			permType: "readfiles",
+			wantPerm: false,
+		},
+		{
+			// A path that is only matched by the allow rule must be permitted.
+			name: "allow_non_denied_path",
+			permissions: []string{
+				"readfiles:!/var/log/secret.*",
+				"readfiles:/var/log/.*",
+			},
+			path:     "/var/log/app.log",
+			permType: "readfiles",
+			wantPerm: true,
+		},
+		{
+			// Multiple deny rules — any matching deny must block the path.
+			name: "multiple_denies_first_matches",
+			permissions: []string{
+				"readfiles:!/var/log/secret.*",
+				"readfiles:!/var/log/private.*",
+				"readfiles:/var/log/.*",
+			},
+			path:     "/var/log/private.log",
+			permType: "readfiles",
+			wantPerm: false,
+		},
+		{
+			// Path not matched by any rule must not gain permission.
+			name: "no_matching_rule_no_permission",
+			permissions: []string{
+				"readfiles:/var/log/.*",
+			},
+			path:     "/etc/passwd",
+			permType: "readfiles",
+			wantPerm: false,
+		},
+		{
+			// Permission type mismatch: rule for a different type must not apply.
+			name: "wrong_permission_type_ignored",
+			permissions: []string{
+				"writefiles:/var/log/.*",
+			},
+			path:     "/var/log/app.log",
+			permType: "readfiles",
+			wantPerm: false,
+		},
+	}
+
+	for _, tc := range tests {
+		tc := tc // capture range var for parallel sub-tests
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			u := newTestUser(tc.permissions)
+			got, err := u.iteratePaths(tc.path, tc.permType)
+			if err != nil {
+				t.Fatalf("iteratePaths returned unexpected error: %v", err)
+			}
+			if got != tc.wantPerm {
+				t.Errorf("iteratePaths(%q, %q) = %v, want %v",
+					tc.path, tc.permType, got, tc.wantPerm)
+			}
+		})
+	}
+}
+
+// TestIteratePaths_InvalidRegex verifies that a malformed regex in the
+// permission list is surfaced as an error rather than silently ignored.
+func TestIteratePaths_InvalidRegex(t *testing.T) {
+	ensureTestDeps(t)
+	t.Parallel()
+
+	u := newTestUser([]string{"readfiles:[invalid"})
+	_, err := u.iteratePaths("/var/log/app.log", "readfiles")
+	if err == nil {
+		t.Error("expected an error for invalid regex, got nil")
+	}
+}
+
+func TestValidateReadTarget_JournalUsesConfigPermissionOnly(t *testing.T) {
+	ensureTestDeps(t)
+	t.Parallel()
+
+	u := newTestUser([]string{`readfiles:^journal:(nginx|postgresql)\.service$`})
+	target, ok := u.ValidateReadTarget("journal:nginx.service", "readfiles")
+	if !ok {
+		t.Fatal("expected journal target to pass config permission")
+	}
+	if target.Kind != fs.JournalKind {
+		t.Fatalf("target.Kind = %v, want %v", target.Kind, fs.JournalKind)
+	}
+}
+
+func TestValidateReadTarget_JournalDeniedByConfigPermission(t *testing.T) {
+	ensureTestDeps(t)
+	t.Parallel()
+
+	u := newTestUser([]string{`readfiles:^journal:(nginx|postgresql)\.service$`})
+	_, ok := u.ValidateReadTarget("journal:ssh.service", "readfiles")
+	if ok {
+		t.Fatal("expected journal target to be denied by config permission")
+	}
+}
+
+func TestValidateReadTarget_JournalDenyRuleWins(t *testing.T) {
+	ensureTestDeps(t)
+	t.Parallel()
+
+	u := newTestUser([]string{
+		`readfiles:!^journal:postgresql\.service$`,
+		`readfiles:^journal:.*\.service$`,
+	})
+	_, ok := u.ValidateReadTarget("journal:postgresql.service", "readfiles")
+	if ok {
+		t.Fatal("expected matching journal deny rule to block target")
+	}
+}