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>

6 files changed, 877 insertions, 50 deletions

diff --git a/internal/ssh/server/authkeystore.go b/internal/ssh/server/authkeystore.go
new file mode 100644
index 0000000..4de71ee
--- /dev/null
+++ b/internal/ssh/server/authkeystore.go
@@ -0,0 +1,186 @@
+package server
+
+import (
+	"sync"
+	"time"
+
+	gossh "golang.org/x/crypto/ssh"
+)
+
+const (
+	defaultAuthKeyTTL        = 24 * time.Hour
+	defaultAuthKeyMaxPerUser = 5
+)
+
+type authKeyEntry struct {
+	pubKey       gossh.PublicKey
+	registeredAt time.Time
+}
+
+// AuthKeyStore is an in-memory, per-user cache of SSH public keys.
+// Each Server instance owns exactly one AuthKeyStore, constructed via
+// NewAuthKeyStore. There is no shared package-level instance; callers
+// must always supply a non-nil store.
+type AuthKeyStore struct {
+	mu             sync.RWMutex
+	keysByUser     map[string][]authKeyEntry
+	ttl            time.Duration
+	maxKeysPerUser int
+	now            func() time.Time
+}
+
+// NewAuthKeyStore builds a thread-safe auth key store.
+func NewAuthKeyStore(ttl time.Duration, maxKeysPerUser int) *AuthKeyStore {
+	return newAuthKeyStoreWithClock(ttl, maxKeysPerUser, time.Now)
+}
+
+func newAuthKeyStoreWithClock(ttl time.Duration, maxKeysPerUser int,
+	nowFn func() time.Time) *AuthKeyStore {
+
+	if ttl <= 0 {
+		ttl = defaultAuthKeyTTL
+	}
+	if maxKeysPerUser <= 0 {
+		maxKeysPerUser = defaultAuthKeyMaxPerUser
+	}
+	if nowFn == nil {
+		nowFn = time.Now
+	}
+
+	return &AuthKeyStore{
+		keysByUser:     make(map[string][]authKeyEntry),
+		ttl:            ttl,
+		maxKeysPerUser: maxKeysPerUser,
+		now:            nowFn,
+	}
+}
+
+// Add stores or refreshes a key for a user.
+func (s *AuthKeyStore) Add(user string, pubKey gossh.PublicKey) {
+	if user == "" || pubKey == nil {
+		return
+	}
+
+	now := s.now()
+	offeredKey := marshalKey(pubKey)
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	userEntries := s.pruneExpiredLocked(user, now)
+
+	newEntries := make([]authKeyEntry, 0, len(userEntries)+1)
+	for _, entry := range userEntries {
+		if marshalKey(entry.pubKey) == offeredKey {
+			continue
+		}
+		newEntries = append(newEntries, entry)
+	}
+
+	newEntries = append(newEntries, authKeyEntry{
+		pubKey:       pubKey,
+		registeredAt: now,
+	})
+	if len(newEntries) > s.maxKeysPerUser {
+		newEntries = newEntries[len(newEntries)-s.maxKeysPerUser:]
+	}
+
+	s.keysByUser[user] = newEntries
+}
+
+// Has returns true if a non-expired key exists for a user.
+func (s *AuthKeyStore) Has(user string, pubKey gossh.PublicKey) bool {
+	if user == "" || pubKey == nil {
+		return false
+	}
+
+	now := s.now()
+	offeredKey := marshalKey(pubKey)
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	userEntries := s.pruneExpiredLocked(user, now)
+	for _, entry := range userEntries {
+		if marshalKey(entry.pubKey) == offeredKey {
+			return true
+		}
+	}
+
+	return false
+}
+
+// Remove deletes a key for a user if it exists.
+func (s *AuthKeyStore) Remove(user string, pubKey gossh.PublicKey) {
+	if user == "" || pubKey == nil {
+		return
+	}
+
+	offeredKey := marshalKey(pubKey)
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	userEntries := s.pruneExpiredLocked(user, s.now())
+	if len(userEntries) == 0 {
+		return
+	}
+
+	remaining := make([]authKeyEntry, 0, len(userEntries))
+	for _, entry := range userEntries {
+		if marshalKey(entry.pubKey) == offeredKey {
+			continue
+		}
+		remaining = append(remaining, entry)
+	}
+
+	if len(remaining) == 0 {
+		delete(s.keysByUser, user)
+		return
+	}
+
+	s.keysByUser[user] = remaining
+}
+
+func (s *AuthKeyStore) pruneExpiredLocked(user string, now time.Time) []authKeyEntry {
+	userEntries, ok := s.keysByUser[user]
+	if !ok || len(userEntries) == 0 {
+		delete(s.keysByUser, user)
+		return nil
+	}
+
+	hasExpiredEntries := false
+	for _, entry := range userEntries {
+		if s.expired(entry, now) {
+			hasExpiredEntries = true
+			break
+		}
+	}
+	if !hasExpiredEntries {
+		return userEntries
+	}
+
+	activeEntries := make([]authKeyEntry, 0, len(userEntries))
+	for _, entry := range userEntries {
+		if s.expired(entry, now) {
+			continue
+		}
+		activeEntries = append(activeEntries, entry)
+	}
+
+	if len(activeEntries) == 0 {
+		delete(s.keysByUser, user)
+		return nil
+	}
+
+	s.keysByUser[user] = activeEntries
+	return activeEntries
+}
+
+func (s *AuthKeyStore) expired(entry authKeyEntry, now time.Time) bool {
+	return !entry.registeredAt.Add(s.ttl).After(now)
+}
+
+func marshalKey(pubKey gossh.PublicKey) string {
+	return string(pubKey.Marshal())
+}
diff --git a/internal/ssh/server/authkeystore_test.go b/internal/ssh/server/authkeystore_test.go
new file mode 100644
index 0000000..056db7b
--- /dev/null
+++ b/internal/ssh/server/authkeystore_test.go
@@ -0,0 +1,178 @@
+package server
+
+import (
+	"crypto/ed25519"
+	"sync"
+	"testing"
+	"time"
+
+	gossh "golang.org/x/crypto/ssh"
+)
+
+func TestAuthKeyStoreAddHasRemove(t *testing.T) {
+	store := NewAuthKeyStore(time.Hour, 5)
+	key := testPublicKey(t, 1)
+
+	if store.Has("alice", key) {
+		t.Fatalf("Store should not contain key before add")
+	}
+
+	store.Add("alice", key)
+	if !store.Has("alice", key) {
+		t.Fatalf("Store should contain key after add")
+	}
+
+	store.Remove("alice", key)
+	if store.Has("alice", key) {
+		t.Fatalf("Store should not contain key after remove")
+	}
+}
+
+func TestAuthKeyStoreHasExpiresKeysLazily(t *testing.T) {
+	now := time.Date(2026, 3, 3, 10, 0, 0, 0, time.UTC)
+	store := newAuthKeyStoreWithClock(10*time.Second, 5, func() time.Time { return now })
+	key := testPublicKey(t, 2)
+
+	store.Add("alice", key)
+	if !store.Has("alice", key) {
+		t.Fatalf("Store should contain fresh key")
+	}
+
+	now = now.Add(11 * time.Second)
+	if store.Has("alice", key) {
+		t.Fatalf("Store should expire key when ttl is exceeded")
+	}
+
+	store.mu.RLock()
+	defer store.mu.RUnlock()
+	if len(store.keysByUser["alice"]) != 0 {
+		t.Fatalf("Expired entries should be removed on Has call")
+	}
+}
+
+func TestAuthKeyStoreEnforcesPerUserKeyLimit(t *testing.T) {
+	now := time.Date(2026, 3, 3, 10, 0, 0, 0, time.UTC)
+	store := newAuthKeyStoreWithClock(time.Hour, 2, func() time.Time { return now })
+
+	keyOne := testPublicKey(t, 3)
+	keyTwo := testPublicKey(t, 4)
+	keyThree := testPublicKey(t, 5)
+
+	store.Add("alice", keyOne)
+	now = now.Add(1 * time.Second)
+	store.Add("alice", keyTwo)
+	now = now.Add(1 * time.Second)
+	store.Add("alice", keyThree)
+
+	if store.Has("alice", keyOne) {
+		t.Fatalf("Oldest key should be evicted once max key limit is reached")
+	}
+	if !store.Has("alice", keyTwo) {
+		t.Fatalf("Second key should remain in store")
+	}
+	if !store.Has("alice", keyThree) {
+		t.Fatalf("Newest key should remain in store")
+	}
+}
+
+func TestAuthKeyStoreAddRefreshesExistingKey(t *testing.T) {
+	now := time.Date(2026, 3, 3, 10, 0, 0, 0, time.UTC)
+	store := newAuthKeyStoreWithClock(10*time.Second, 5, func() time.Time { return now })
+	key := testPublicKey(t, 6)
+
+	store.Add("alice", key)
+	now = now.Add(9 * time.Second)
+	store.Add("alice", key)
+
+	now = now.Add(5 * time.Second)
+	if !store.Has("alice", key) {
+		t.Fatalf("Key should stay valid after it is refreshed")
+	}
+
+	now = now.Add(6 * time.Second)
+	if store.Has("alice", key) {
+		t.Fatalf("Refreshed key should expire once ttl is exceeded from latest add")
+	}
+}
+
+func TestAuthKeyStoreUserIsolation(t *testing.T) {
+	store := NewAuthKeyStore(time.Hour, 5)
+	key := testPublicKey(t, 7)
+
+	store.Add("alice", key)
+	if store.Has("bob", key) {
+		t.Fatalf("Key lookup must be isolated by user")
+	}
+}
+
+func TestAuthKeyStoreIgnoresInvalidInput(t *testing.T) {
+	store := NewAuthKeyStore(time.Hour, 5)
+	key := testPublicKey(t, 8)
+
+	store.Add("", key)
+	store.Add("alice", nil)
+	store.Remove("", key)
+	store.Remove("alice", nil)
+
+	if store.Has("", key) {
+		t.Fatalf("Empty user should not match")
+	}
+	if store.Has("alice", nil) {
+		t.Fatalf("Nil key should not match")
+	}
+}
+
+func TestAuthKeyStoreConcurrentAccess(t *testing.T) {
+	store := NewAuthKeyStore(time.Hour, 5)
+	users := []string{"alice", "bob", "carol"}
+	keys := []gossh.PublicKey{
+		testPublicKey(t, 11),
+		testPublicKey(t, 12),
+		testPublicKey(t, 13),
+		testPublicKey(t, 14),
+	}
+
+	var wg sync.WaitGroup
+	for worker := 0; worker < 32; worker++ {
+		wg.Add(1)
+		go func(workerID int) {
+			defer wg.Done()
+
+			user := users[workerID%len(users)]
+			for i := 0; i < 200; i++ {
+				key := keys[(workerID+i)%len(keys)]
+				store.Add(user, key)
+				_ = store.Has(user, key)
+				if i%3 == 0 {
+					store.Remove(user, key)
+				}
+			}
+		}(worker)
+	}
+	wg.Wait()
+
+	store.mu.RLock()
+	defer store.mu.RUnlock()
+	for user, userEntries := range store.keysByUser {
+		if len(userEntries) > store.maxKeysPerUser {
+			t.Fatalf("User %s exceeded max key limit: %d", user, len(userEntries))
+		}
+	}
+}
+
+func testPublicKey(t *testing.T, seedByte byte) gossh.PublicKey {
+	t.Helper()
+
+	seed := make([]byte, ed25519.SeedSize)
+	for i := range seed {
+		seed[i] = seedByte
+	}
+
+	privateKey := ed25519.NewKeyFromSeed(seed)
+	publicKey, err := gossh.NewPublicKey(privateKey.Public())
+	if err != nil {
+		t.Fatalf("Unable to build ssh public key: %s", err.Error())
+	}
+
+	return publicKey
+}
diff --git a/internal/ssh/server/hostkey.go b/internal/ssh/server/hostkey.go
index b2d4569..1315351 100644
--- a/internal/ssh/server/hostkey.go
+++ b/internal/ssh/server/hostkey.go
@@ -1,41 +1,75 @@
 package server
 
 import (
-	"os"
+	"errors"
+	iofs "io/fs"
 
 	"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/ssh"
 )
 
+const (
+	defaultHostKeyBits = 4096
+	defaultHostKeyFile = "./cache/ssh_host_key"
+)
+
 // PrivateHostKey retrieves the private server RSA host key.
-func PrivateHostKey() []byte {
-	hostKeyFile := config.Server.HostKeyFile
+func PrivateHostKey(hostKeyFile string, hostKeyBits int) []byte {
+	if hostKeyFile == "" {
+		hostKeyFile = defaultHostKeyFile
+	}
+	if hostKeyBits <= 0 {
+		hostKeyBits = defaultHostKeyBits
+	}
 	if config.Env("DTAIL_INTEGRATION_TEST_RUN_MODE") {
 		hostKeyFile = "./ssh_host_key"
 	}
-	_, err := os.Stat(hostKeyFile)
-
-	if os.IsNotExist(err) {
-		dlog.Server.Info("Generating private server RSA host key")
-		privateKey, err := ssh.GeneratePrivateRSAKey(config.Server.HostKeyBits)
-
-		if err != nil {
-			dlog.Server.FatalPanic("Failed to generate private server RSA host key", err)
-		}
+	hostKeyPath, err := fs.NewRootedPath(hostKeyFile)
+	if err != nil {
+		dlog.Server.FatalPanic("Invalid private server RSA host key path", hostKeyFile, err)
+	}
 
-		pem := ssh.EncodePrivateKeyToPEM(privateKey)
-		if err := os.WriteFile(hostKeyFile, pem, 0600); err != nil {
-			dlog.Server.Error("Unable to write private server RSA host key to file",
-				hostKeyFile, err)
+	_, err = hostKeyPath.Stat()
+	if err != nil {
+		// os.IsNotExist does not unwrap fmt.Errorf chains from RootedPath.Stat; use errors.Is.
+		if errors.Is(err, iofs.ErrNotExist) {
+			dlog.Server.Info("Generating private server RSA host key")
+			pem, genErr := generatePrivateHostKey(hostKeyBits)
+			if genErr != nil {
+				dlog.Server.FatalPanic("Failed to generate private server RSA host key", genErr)
+			}
+			if storeErr := storePrivateHostKey(hostKeyPath, pem); storeErr != nil {
+				dlog.Server.Error("Unable to write private server RSA host key to file",
+					hostKeyFile, storeErr)
+			}
+			return pem
 		}
-		return pem
+		dlog.Server.FatalPanic("Cannot stat private server RSA host key path", hostKeyFile, err)
 	}
 
 	dlog.Server.Info("Reading private server RSA host key from file", hostKeyFile)
-	pem, err := os.ReadFile(hostKeyFile)
+	pem, err := readPrivateHostKey(hostKeyPath)
 	if err != nil {
 		dlog.Server.FatalPanic("Failed to load private server RSA host key", err)
 	}
 	return pem
 }
+
+func generatePrivateHostKey(hostKeyBits int) ([]byte, error) {
+	privateKey, err := ssh.GeneratePrivateRSAKey(hostKeyBits)
+	if err != nil {
+		return nil, err
+	}
+
+	return ssh.EncodePrivateKeyToPEM(privateKey), nil
+}
+
+func storePrivateHostKey(hostKeyPath fs.RootedPath, pem []byte) error {
+	return hostKeyPath.WriteFile(pem, 0o600)
+}
+
+func readPrivateHostKey(hostKeyPath fs.RootedPath) ([]byte, error) {
+	return hostKeyPath.ReadFile()
+}
diff --git a/internal/ssh/server/hostkey_test.go b/internal/ssh/server/hostkey_test.go
new file mode 100644
index 0000000..e318fb4
--- /dev/null
+++ b/internal/ssh/server/hostkey_test.go
@@ -0,0 +1,37 @@
+package server
+
+import (
+	"bytes"
+	"github.com/mimecast/dtail/internal/io/fs"
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func TestPrivateHostKeyGeneratesAndReloadsExistingKey(t *testing.T) {
+	hostKeyFile := filepath.Join(t.TempDir(), "cache", "ssh_host_key")
+	if err := os.MkdirAll(filepath.Dir(hostKeyFile), 0o755); err != nil {
+		t.Fatalf("MkdirAll failed: %v", err)
+	}
+
+	hostKeyPath, err := fs.NewRootedPath(hostKeyFile)
+	if err != nil {
+		t.Fatalf("NewRootedPath failed: %v", err)
+	}
+
+	firstPEM, err := generatePrivateHostKey(1024)
+	if err != nil {
+		t.Fatalf("generatePrivateHostKey failed: %v", err)
+	}
+	if err := storePrivateHostKey(hostKeyPath, firstPEM); err != nil {
+		t.Fatalf("storePrivateHostKey failed: %v", err)
+	}
+
+	secondPEM, err := readPrivateHostKey(hostKeyPath)
+	if err != nil {
+		t.Fatalf("readPrivateHostKey failed: %v", err)
+	}
+	if !bytes.Equal(secondPEM, firstPEM) {
+		t.Fatalf("readPrivateHostKey returned different key data")
+	}
+}
diff --git a/internal/ssh/server/publickeycallback.go b/internal/ssh/server/publickeycallback.go
index bcc9004..8a23384 100644
--- a/internal/ssh/server/publickeycallback.go
+++ b/internal/ssh/server/publickeycallback.go
@@ -1,38 +1,64 @@
 package server
 
 import (
+	"bytes"
+	"errors"
 	"fmt"
+	iofs "io/fs"
 	"os"
 	goUser "os/user"
+	"path/filepath"
 
 	"github.com/mimecast/dtail/internal/config"
 	"github.com/mimecast/dtail/internal/io/dlog"
+	"github.com/mimecast/dtail/internal/io/fs"
 	user "github.com/mimecast/dtail/internal/user/server"
 
 	gossh "golang.org/x/crypto/ssh"
 )
 
-// PublicKeyCallback is for the server to check whether a public SSH key is
-// authorized ot not.
-func PublicKeyCallback(c gossh.ConnMetadata,
-	offeredPubKey gossh.PublicKey) (*gossh.Permissions, error) {
+type authorizedKeyParser func([]byte) (gossh.PublicKey, string, []string, []byte, error)
+
+// NewPublicKeyCallback creates an instance-scoped SSH public key callback.
+// keyStore must be non-nil; callers are responsible for constructing and
+// wiring the store. There is no shared package-level fallback.
+func NewPublicKeyCallback(authKeyEnabled bool, cacheDir string,
+	keyStore *AuthKeyStore) func(gossh.ConnMetadata, gossh.PublicKey) (*gossh.Permissions, error) {
 
-	user, err := user.New(c.User(), c.RemoteAddr().String())
+	if keyStore == nil {
+		panic("NewPublicKeyCallback: keyStore must not be nil")
+	}
+	return func(c gossh.ConnMetadata, offeredPubKey gossh.PublicKey) (*gossh.Permissions, error) {
+		return publicKeyCallback(c, offeredPubKey, authKeyEnabled, cacheDir, keyStore)
+	}
+}
+
+func publicKeyCallback(c gossh.ConnMetadata, offeredPubKey gossh.PublicKey,
+	authKeyEnabled bool, cacheDir string, keyStore *AuthKeyStore) (*gossh.Permissions, error) {
+
+	user, err := user.New(c.User(), c.RemoteAddr().String(), nil)
 	if err != nil {
 		return nil, err
 	}
 	dlog.Server.Info(user, "Incoming authorization")
 
-	authorizedKeysFile, err := authorizedKeysFile(user)
+	if authKeyEnabled {
+		if permissions := authKeyStorePermissions(keyStore, user.Name, offeredPubKey); permissions != nil {
+			dlog.Server.Info(user, "Authorized by in-memory auth key store")
+			return permissions, nil
+		}
+	}
+
+	authorizedKeysPath, err := authorizedKeysPathForUser(user, cacheDir)
 	if err != nil {
 		return nil, err
 	}
 
-	dlog.Server.Info(user, "Reading", authorizedKeysFile)
-	authorizedKeysBytes, err := os.ReadFile(authorizedKeysFile)
+	dlog.Server.Info(user, "Reading", authorizedKeysPath.Path())
+	authorizedKeysBytes, err := authorizedKeysPath.ReadFile()
 	if err != nil {
 		return nil, fmt.Errorf("Unable to read authorized keys file|%s|%s|%s",
-			authorizedKeysFile, user, err.Error())
+			authorizedKeysPath.Path(), user, err.Error())
 	}
 
 	return verifyAuthorizedKeys(user, authorizedKeysBytes, offeredPubKey)
@@ -40,57 +66,126 @@ func PublicKeyCallback(c gossh.ConnMetadata,
 
 func verifyAuthorizedKeys(user *user.User, authorizedKeysBytes []byte,
 	offeredPubKey gossh.PublicKey) (*gossh.Permissions, error) {
+	return verifyAuthorizedKeysWithParser(user, authorizedKeysBytes, offeredPubKey, gossh.ParseAuthorizedKey)
+}
+
+func verifyAuthorizedKeysWithParser(user *user.User, authorizedKeysBytes []byte,
+	offeredPubKey gossh.PublicKey, parseAuthorizedKey authorizedKeyParser) (*gossh.Permissions, error) {
 
 	authorizedKeysMap := map[string]bool{}
 	for len(authorizedKeysBytes) > 0 {
-		authorizedPubKey, _, _, restBytes, err := gossh.ParseAuthorizedKey(authorizedKeysBytes)
+		authorizedPubKey, _, _, restBytes, err := parseAuthorizedKey(authorizedKeysBytes)
 		if err != nil {
-			return nil, fmt.Errorf("unable to parse authorized keys bytes|%s|%s",
-				user, err.Error())
+			if dlog.Server != nil {
+				dlog.Server.Warn(user, "Skipping unparseable authorized_keys line", err)
+			}
+			nextAuthorizedKeysBytes, ok := advanceToNextAuthorizedKeysLine(authorizedKeysBytes)
+			if !ok {
+				break
+			}
+			authorizedKeysBytes = nextAuthorizedKeysBytes
+			continue
 		}
 		authorizedKeysMap[string(authorizedPubKey.Marshal())] = true
 		authorizedKeysBytes = restBytes
-		dlog.Server.Debug(user, "Authorized public key fingerprint",
-			gossh.FingerprintSHA256(authorizedPubKey))
+		if dlog.Server != nil {
+			dlog.Server.Debug(user, "Authorized public key fingerprint",
+				gossh.FingerprintSHA256(authorizedPubKey))
+		}
 	}
 
-	dlog.Server.Debug(user, "Offered public key fingerprint", gossh.FingerprintSHA256(offeredPubKey))
+	if dlog.Server != nil {
+		dlog.Server.Debug(user, "Offered public key fingerprint", gossh.FingerprintSHA256(offeredPubKey))
+	}
 	if authorizedKeysMap[string(offeredPubKey.Marshal())] {
-		return &gossh.Permissions{
-			Extensions: map[string]string{"pubkey-fp": gossh.FingerprintSHA256(offeredPubKey)},
-		}, nil
+		return permissionsFromPublicKey(offeredPubKey), nil
 	}
 
 	return nil, fmt.Errorf("%s|public key of user not authorized", user)
 }
 
-func authorizedKeysFile(user *user.User) (string, error) {
+func advanceToNextAuthorizedKeysLine(authorizedKeysBytes []byte) ([]byte, bool) {
+	lineEnd := bytes.IndexByte(authorizedKeysBytes, '\n')
+	if lineEnd == -1 {
+		return nil, false
+	}
+
+	nextBytes := authorizedKeysBytes[lineEnd+1:]
+	return nextBytes, true
+}
+
+func authKeyStorePermissions(keyStore *AuthKeyStore, userName string,
+	offeredPubKey gossh.PublicKey) *gossh.Permissions {
+
+	if keyStore == nil || !keyStore.Has(userName, offeredPubKey) {
+		return nil
+	}
+
+	return permissionsFromPublicKey(offeredPubKey)
+}
+
+func permissionsFromPublicKey(offeredPubKey gossh.PublicKey) *gossh.Permissions {
+	return &gossh.Permissions{
+		Extensions: map[string]string{"pubkey-fp": gossh.FingerprintSHA256(offeredPubKey)},
+	}
+}
+
+type userLookupFunc func(string) (*goUser.User, error)
+
+func authorizedKeysPathForUser(user *user.User, cacheDir string) (fs.RootedPath, error) {
 	if config.Env("DTAIL_INTEGRATION_TEST_RUN_MODE") {
 		// In this case, we expect a pub key in the current directory.
-		return "./id_rsa.pub", nil
+		return fs.NewRootedPath("./id_rsa.pub")
 	}
 
 	cwd, err := os.Getwd()
 	if err != nil {
-		return "", err
+		return fs.RootedPath{}, err
 	}
 
-	// Check for cached version in the dserver directory.
-	authorizedKeysFile := fmt.Sprintf("%s/%s/%s.authorized_keys", cwd,
-		config.Common.CacheDir, user.Name)
-	if _, err = os.Stat(authorizedKeysFile); err == nil {
-		return authorizedKeysFile, nil
+	return findAuthorizedKeysPath(user, cacheDir, cwd, goUser.Lookup)
+}
+
+func findAuthorizedKeysPath(user *user.User, cacheDir, cwd string,
+	lookupUser userLookupFunc) (fs.RootedPath, error) {
+
+	// Check for cached version in the dserver directory. An absolute
+	// CacheDir (as used by the BSD packages, e.g. /var/run/dserver/cache)
+	// must be used as-is: joining it with the CWD would break auth
+	// whenever dserver is started from a directory other than / (e.g. a
+	// manual rc.d restart from a home directory). Relative CacheDirs stay
+	// relative to the CWD as before.
+	if cacheDir != "" {
+		cacheBase := cacheDir
+		if !filepath.IsAbs(cacheDir) {
+			cacheBase = filepath.Join(cwd, cacheDir)
+		}
+		cachePath := filepath.Join(cacheBase, fmt.Sprintf("%s.authorized_keys", user.Name))
+		rootedCachePath, err := fs.NewRootedPath(cachePath)
+		if err != nil {
+			return fs.RootedPath{}, err
+		}
+		if _, err := rootedCachePath.Stat(); err == nil {
+			return rootedCachePath, nil
+		}
 	}
 
 	// As the last option, check the regular SSH path.
-	osUser, err := goUser.Lookup(user.Name)
+	osUser, err := lookupUser(user.Name)
+	if err != nil {
+		return fs.RootedPath{}, err
+	}
+	authorizedKeysPath := filepath.Join(osUser.HomeDir, ".ssh", "authorized_keys")
+	rootedAuthorizedKeysPath, err := fs.NewRootedPath(authorizedKeysPath)
 	if err != nil {
-		return "", err
+		return fs.RootedPath{}, err
+	}
+	if _, err = rootedAuthorizedKeysPath.Stat(); err == nil {
+		return rootedAuthorizedKeysPath, nil
 	}
-	authorizedKeysFile = fmt.Sprintf("%s/.ssh/authorized_keys", osUser.HomeDir)
-	if _, err = os.Stat(authorizedKeysFile); err == nil {
-		return authorizedKeysFile, nil
+	if !errors.Is(err, iofs.ErrNotExist) {
+		return fs.RootedPath{}, err
 	}
 
-	return "", fmt.Errorf("unable to find a any authorized keys file")
+	return fs.RootedPath{}, fmt.Errorf("unable to find any authorized keys file")
 }
diff --git a/internal/ssh/server/publickeycallback_test.go b/internal/ssh/server/publickeycallback_test.go
new file mode 100644
index 0000000..2f597cb
--- /dev/null
+++ b/internal/ssh/server/publickeycallback_test.go
@@ -0,0 +1,297 @@
+package server
+
+import (
+	"bytes"
+	"errors"
+	"os"
+	goUser "os/user"
+	"path/filepath"
+	"testing"
+	"time"
+
+	serveruser "github.com/mimecast/dtail/internal/user/server"
+
+	gossh "golang.org/x/crypto/ssh"
+)
+
+func TestAuthKeyStorePermissions(t *testing.T) {
+	// Create an isolated store for this test — there is no package-level global.
+	store := NewAuthKeyStore(time.Hour, 5)
+
+	key := testPublicKey(t, 21)
+
+	if permissions := authKeyStorePermissions(store, "alice", key); permissions != nil {
+		t.Fatalf("Expected nil permissions when no key is cached")
+	}
+
+	store.Add("alice", key)
+
+	permissions := authKeyStorePermissions(store, "alice", key)
+	if permissions == nil {
+		t.Fatalf("Expected permissions when key is cached")
+	}
+	if fingerprint := permissions.Extensions["pubkey-fp"]; fingerprint != gossh.FingerprintSHA256(key) {
+		t.Fatalf("Unexpected fingerprint: %s", fingerprint)
+	}
+
+	if permissions := authKeyStorePermissions(store, "bob", key); permissions != nil {
+		t.Fatalf("Expected nil permissions for different user")
+	}
+
+	unknownKey := testPublicKey(t, 22)
+	if permissions := authKeyStorePermissions(store, "alice", unknownKey); permissions != nil {
+		t.Fatalf("Expected nil permissions for unknown key")
+	}
+}
+
+func TestVerifyAuthorizedKeysSkipsMalformedLineWithoutParserProgress(t *testing.T) {
+	user := testServerUser(t, "alice")
+	firstKey := testPublicKey(t, 41)
+	secondKey := testPublicKey(t, 42)
+
+	firstLine := gossh.MarshalAuthorizedKey(firstKey)
+	badLine := []byte("this is not an authorized key\n")
+	secondLine := gossh.MarshalAuthorizedKey(secondKey)
+	authorizedKeys := append(append(append([]byte{}, firstLine...), badLine...), secondLine...)
+
+	parser := func(in []byte) (gossh.PublicKey, string, []string, []byte, error) {
+		switch {
+		case bytes.HasPrefix(in, firstLine):
+			return firstKey, "", nil, in[len(firstLine):], nil
+		case bytes.HasPrefix(in, badLine):
+			return nil, "", nil, in, errors.New("parse error")
+		case bytes.HasPrefix(in, secondLine):
+			return secondKey, "", nil, in[len(secondLine):], nil
+		default:
+			return nil, "", nil, nil, errors.New("unexpected authorized_keys input")
+		}
+	}
+
+	permissions, err := verifyAuthorizedKeysWithParser(user, authorizedKeys, secondKey, parser)
+	if err != nil {
+		t.Fatalf("verifyAuthorizedKeysWithParser failed: %v", err)
+	}
+	if permissions == nil {
+		t.Fatalf("Expected permissions for key after malformed line")
+	}
+	if got := permissions.Extensions["pubkey-fp"]; got != gossh.FingerprintSHA256(secondKey) {
+		t.Fatalf("Unexpected fingerprint: %s", got)
+	}
+}
+
+func TestVerifyAuthorizedKeysSkipsMalformedLineWithRealParser(t *testing.T) {
+	user := testServerUser(t, "alice")
+	firstKey := testPublicKey(t, 43)
+	secondKey := testPublicKey(t, 44)
+
+	badLine := []byte("ssh-rsa !!!!\n")
+	authorizedKeys := append(append(append([]byte{}, gossh.MarshalAuthorizedKey(firstKey)...),
+		badLine...), gossh.MarshalAuthorizedKey(secondKey)...)
+
+	sawParseError := false
+	parseAuthorizedKeyLineByLine := func(in []byte) (gossh.PublicKey, string, []string, []byte, error) {
+		line := in
+		rest := []byte(nil)
+		if lineEnd := bytes.IndexByte(in, '\n'); lineEnd >= 0 {
+			line = in[:lineEnd+1]
+			rest = in[lineEnd+1:]
+		}
+
+		authorizedPubKey, comment, options, _, err := gossh.ParseAuthorizedKey(line)
+		if err != nil {
+			if bytes.Equal(line, badLine) {
+				sawParseError = true
+			}
+			return nil, "", nil, rest, err
+		}
+
+		return authorizedPubKey, comment, options, rest, nil
+	}
+
+	permissions, err := verifyAuthorizedKeysWithParser(user, authorizedKeys, secondKey, parseAuthorizedKeyLineByLine)
+	if err != nil {
+		t.Fatalf("verifyAuthorizedKeysWithParser failed: %v", err)
+	}
+	if permissions == nil {
+		t.Fatalf("Expected permissions for key after malformed line")
+	}
+	if got := permissions.Extensions["pubkey-fp"]; got != gossh.FingerprintSHA256(secondKey) {
+		t.Fatalf("Unexpected fingerprint: %s", got)
+	}
+	if !sawParseError {
+		t.Fatalf("Expected malformed authorized_keys line to hit the real parser error path")
+	}
+}
+
+func TestFindAuthorizedKeysPathUsesCacheDirWhenPresent(t *testing.T) {
+	cwd := t.TempDir()
+	cacheDir := "cache"
+	user := testServerUser(t, "alice")
+	wantPath := filepath.Join(cwd, cacheDir, "alice.authorized_keys")
+	if err := os.MkdirAll(filepath.Dir(wantPath), 0o755); err != nil {
+		t.Fatalf("MkdirAll failed: %v", err)
+	}
+
+	want := gossh.MarshalAuthorizedKey(testPublicKey(t, 31))
+	if err := os.WriteFile(wantPath, want, 0o600); err != nil {
+		t.Fatalf("WriteFile failed: %v", err)
+	}
+
+	rootedPath, err := findAuthorizedKeysPath(user, cacheDir, cwd, func(string) (*goUser.User, error) {
+		t.Fatalf("lookupUser should not be called when cached authorized_keys exists")
+		return nil, nil
+	})
+	if err != nil {
+		t.Fatalf("findAuthorizedKeysPath failed: %v", err)
+	}
+	if rootedPath.Path() != wantPath {
+		t.Fatalf("findAuthorizedKeysPath returned %q, want %q", rootedPath.Path(), wantPath)
+	}
+
+	got, err := rootedPath.ReadFile()
+	if err != nil {
+		t.Fatalf("ReadFile failed: %v", err)
+	}
+	if !bytes.Equal(got, want) {
+		t.Fatalf("ReadFile returned %q, want %q", got, want)
+	}
+}
+
+func TestFindAuthorizedKeysPathIgnoresCwdForAbsoluteCacheDir(t *testing.T) {
+	// An absolute cache dir (e.g. /var/run/dserver/cache on the BSD
+	// packages) must resolve independently of the CWD dserver was
+	// started from.
+	cwd := filepath.Join(t.TempDir(), "unrelated-cwd")
+	cacheDir := t.TempDir()
+	user := testServerUser(t, "alice")
+	wantPath := filepath.Join(cacheDir, "alice.authorized_keys")
+
+	want := gossh.MarshalAuthorizedKey(testPublicKey(t, 33))
+	if err := os.WriteFile(wantPath, want, 0o600); err != nil {
+		t.Fatalf("WriteFile failed: %v", err)
+	}
+
+	rootedPath, err := findAuthorizedKeysPath(user, cacheDir, cwd, func(string) (*goUser.User, error) {
+		t.Fatalf("lookupUser should not be called when cached authorized_keys exists")
+		return nil, nil
+	})
+	if err != nil {
+		t.Fatalf("findAuthorizedKeysPath failed: %v", err)
+	}
+	if rootedPath.Path() != wantPath {
+		t.Fatalf("findAuthorizedKeysPath returned %q, want %q", rootedPath.Path(), wantPath)
+	}
+
+	got, err := rootedPath.ReadFile()
+	if err != nil {
+		t.Fatalf("ReadFile failed: %v", err)
+	}
+	if !bytes.Equal(got, want) {
+		t.Fatalf("ReadFile returned %q, want %q", got, want)
+	}
+}
+
+func TestFindAuthorizedKeysPathAbsoluteCacheDirMissingFileFallsBack(t *testing.T) {
+	// An absolute cache dir without a per-user cache file must fall back
+	// to ~/.ssh/authorized_keys instead of erroring out.
+	cwd := t.TempDir()
+	cacheDir := t.TempDir() // exists, but holds no alice.authorized_keys
+	homeDir := t.TempDir()
+	user := testServerUser(t, "alice")
+	wantPath := filepath.Join(homeDir, ".ssh", "authorized_keys")
+	if err := os.MkdirAll(filepath.Dir(wantPath), 0o755); err != nil {
+		t.Fatalf("MkdirAll failed: %v", err)
+	}
+
+	want := gossh.MarshalAuthorizedKey(testPublicKey(t, 34))
+	if err := os.WriteFile(wantPath, want, 0o600); err != nil {
+		t.Fatalf("WriteFile failed: %v", err)
+	}
+
+	rootedPath, err := findAuthorizedKeysPath(user, cacheDir, cwd, func(name string) (*goUser.User, error) {
+		return &goUser.User{Username: name, HomeDir: homeDir}, nil
+	})
+	if err != nil {
+		t.Fatalf("findAuthorizedKeysPath failed: %v", err)
+	}
+	if rootedPath.Path() != wantPath {
+		t.Fatalf("findAuthorizedKeysPath returned %q, want %q", rootedPath.Path(), wantPath)
+	}
+
+	got, err := rootedPath.ReadFile()
+	if err != nil {
+		t.Fatalf("ReadFile failed: %v", err)
+	}
+	if !bytes.Equal(got, want) {
+		t.Fatalf("ReadFile returned %q, want %q", got, want)
+	}
+}
+
+func TestFindAuthorizedKeysPathFallsBackToHomeAuthorizedKeys(t *testing.T) {
+	cwd := t.TempDir()
+	homeDir := t.TempDir()
+	user := testServerUser(t, "alice")
+	wantPath := filepath.Join(homeDir, ".ssh", "authorized_keys")
+	if err := os.MkdirAll(filepath.Dir(wantPath), 0o755); err != nil {
+		t.Fatalf("MkdirAll failed: %v", err)
+	}
+
+	want := gossh.MarshalAuthorizedKey(testPublicKey(t, 32))
+	if err := os.WriteFile(wantPath, want, 0o600); err != nil {
+		t.Fatalf("WriteFile failed: %v", err)
+	}
+
+	rootedPath, err := findAuthorizedKeysPath(user, "cache", cwd, func(name string) (*goUser.User, error) {
+		return &goUser.User{Username: name, HomeDir: homeDir}, nil
+	})
+	if err != nil {
+		t.Fatalf("findAuthorizedKeysPath failed: %v", err)
+	}
+	if rootedPath.Path() != wantPath {
+		t.Fatalf("findAuthorizedKeysPath returned %q, want %q", rootedPath.Path(), wantPath)
+	}
+
+	got, err := rootedPath.ReadFile()
+	if err != nil {
+		t.Fatalf("ReadFile failed: %v", err)
+	}
+	if !bytes.Equal(got, want) {
+		t.Fatalf("ReadFile returned %q, want %q", got, want)
+	}
+}
+
+func TestFindAuthorizedKeysPathRejectsEscapingHomeSymlink(t *testing.T) {
+	cwd := t.TempDir()
+	homeDir := t.TempDir()
+	user := testServerUser(t, "alice")
+	sshDir := filepath.Join(homeDir, ".ssh")
+	if err := os.MkdirAll(sshDir, 0o755); err != nil {
+		t.Fatalf("MkdirAll failed: %v", err)
+	}
+
+	outsidePath := filepath.Join(homeDir, "outside_authorized_keys")
+	if err := os.WriteFile(outsidePath, gossh.MarshalAuthorizedKey(testPublicKey(t, 33)), 0o600); err != nil {
+		t.Fatalf("WriteFile failed: %v", err)
+	}
+	if err := os.Symlink(filepath.Join("..", "outside_authorized_keys"),
+		filepath.Join(sshDir, "authorized_keys")); err != nil {
+		t.Fatalf("Symlink failed: %v", err)
+	}
+
+	_, err := findAuthorizedKeysPath(user, "", cwd, func(name string) (*goUser.User, error) {
+		return &goUser.User{Username: name, HomeDir: homeDir}, nil
+	})
+	if err == nil {
+		t.Fatalf("findAuthorizedKeysPath succeeded for escaping authorized_keys symlink")
+	}
+}
+
+func testServerUser(t *testing.T, name string) *serveruser.User {
+	t.Helper()
+
+	user, err := serveruser.New(name, "127.0.0.1:2222", nil)
+	if err != nil {
+		t.Fatalf("serveruser.New failed: %v", err)
+	}
+	return user
+}