main: Exit if input files are not named pipes.

In discussions with the team, we decided that the input files will
always be named pipes. As a result, we felt it was necessary to
enforce a file-type check that produces a helpful error message.

cmd/namedpipe.go

@@ -136,6 +136,12 @@ func RunNamedPipe(ctx context.Context, osArgs []string, h *health.Health, optLog
 
 	h.AddReadiness(namedpipe.NamedPipeProcessorComponentName)
 	eg.Go(func() error {
+		err := common.IsNamedPipe(sshdLogFilePath)
+		if err != nil {
+			return fmt.Errorf("failed to check if sshd log path is a named pipe: %q - %w",
+				sshdLogFilePath, err)
+		}
+
 		sshdProcessor := sshd.NewSshdProcessor(groupCtx, logins, nodeName, mid, eventWriter, pprov)
 		npi := namedpipe.NewNamedPipeIngester(logger, h)
 		if distro == util.DistroRocky {
@@ -158,10 +164,16 @@ func RunNamedPipe(ctx context.Context, osArgs []string, h *health.Health, optLog
 
 	h.AddReadiness(namedpipe.NamedPipeProcessorComponentName)
 	eg.Go(func() error {
+		err := common.IsNamedPipe(auditdLogFilePath)
+		if err != nil {
+			return fmt.Errorf("failed to check if auditd log path is a named pipe: %q - %w",
+				auditdLogFilePath, err)
+		}
+
 		np := namedpipe.NewNamedPipeIngester(logger, h)
 		alp := auditlog.NewAuditLogIngester(auditdLogFilePath, auditLogChan, np)
 
-		err := alp.Ingest(groupCtx)
+		err = alp.Ingest(groupCtx)
 		if logger.Level().Enabled(zap.DebugLevel) {
 			logger.Debugf("audit log ingester exited (%v)", err)
 		}

internal/common/namedpipe.go

@@ -0,0 +1,23 @@
+package common
+
+import (
+	"errors"
+	"os"
+)
+
+var errNotNamedPipe = errors.New("not a named pipe")
+
+// IsNamedPipe returns a non-nil error if filePath cannot be stat'ed
+// or if it is not a named pipe.
+func IsNamedPipe(filePath string) error {
+	sshdLogFileInfo, err := os.Stat(filePath)
+	if err != nil {
+		return err
+	}
+
+	if (sshdLogFileInfo.Mode() & os.ModeNamedPipe) == os.ModeNamedPipe {
+		return nil
+	}
+
+	return errNotNamedPipe
+}

internal/common/namedpipe_test.go

@@ -0,0 +1,42 @@
+package common
+
+import (
+	"os"
+	"path/filepath"
+	"syscall"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestIsNamedPipe(t *testing.T) {
+	t.Parallel()
+
+	namedPipePath := filepath.Join(t.TempDir(), "foo.pipe")
+
+	require.NoError(t, syscall.Mkfifo(namedPipePath, 0o600))
+
+	require.NoError(t, IsNamedPipe(namedPipePath))
+}
+
+func TestIsNamedPipe_RegularFile(t *testing.T) {
+	t.Parallel()
+
+	regularFilePath := filepath.Join(t.TempDir(), "foo.txt")
+
+	regularFile, err := os.Create(regularFilePath)
+	require.NoError(t, err)
+	_ = regularFile.Close()
+
+	require.ErrorIs(t, IsNamedPipe(regularFilePath), errNotNamedPipe)
+}
+
+func TestIsNamedPipe_StatFailure(t *testing.T) {
+	t.Parallel()
+
+	regularFilePath := filepath.Join(t.TempDir(), string([]byte{0x90, 0x90, 0x90, 0x90}))
+
+	var expErr *os.PathError
+
+	require.ErrorAs(t, IsNamedPipe(regularFilePath), &expErr)
+}