logs: enable NoFollow for process_log-discovered paths

Turn on the no-follow enforcement plumbed through in the previous stack
entries, for the one provider it's meant for:

- pkg/logs/schedulers/ad/scheduler.go sets cfg.NoFollow = true for sources
  whose config.Provider is names.ProcessLog. process_log paths come from
  readlink(/proc/<pid>/fd/<n>); the kernel resolves all symlinks at
  file-open time, so the string in /proc/fd is already canonical - any
  symlink appearing later at that path was planted after discovery and
  indicates an attacker-controlled swap. Other providers (file, kubernetes,
  ...) are unaffected: their paths are explicitly specified by the user, and
  it's up to the user to ensure such a path isn't swapped for a symlink by
  an untrusted party.
- comp/core/autodiscovery/providers/process_log.go's checkFileReadable now
  opens with privilegedlogsclient.OpenNoFollow instead of Open, to match
  what the tailer does when it actually reads the discovered file.

Closes DSCVR-475. Final PR in the stack split out of PR #51746 for easier
review; depends on the tailer/fingerprinter NoFollow-plumbing PR.

Environment: Datadog workspace

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Co-authored-by: Vincent Whitchurch <vincent.whitchurch@datadoghq.com>

Author: vitkyrka

comp/core/autodiscovery/providers/process_log.go

@@ -219,7 +219,7 @@ func checkFileReadable(logPath string) error {
 	// Check readability with the privileged logs client to match what the
 	// log tailer uses.  That client can use the privileged logs module in
 	// system-probe if it is available.
-	file, err := privilegedlogsclient.Open(logPath)
+	file, err := privilegedlogsclient.OpenNoFollow(logPath)
 	if err != nil {
 		log.Infof("Discovered log file %s could not be opened: %v", logPath, err)
 		return err

pkg/logs/schedulers/ad/scheduler.go

@@ -262,6 +262,14 @@ func CreateSources(config integration.Config) ([]*sourcesPkg.LogSource, error) {
 
 		cfg.IntegrationSourceIndex = index
 		cfg.IntegrationSource = config.Source
+		if config.Provider == names.ProcessLog {
+			// process_log paths come from readlink(/proc/<pid>/fd/<n>).  The Linux
+			// kernel resolves all symlinks at file-open time, so the string in
+			// /proc/fd is already canonical — any symlink that appears later was
+			// planted after discovery and indicates an attacker-controlled swap.
+			// NoFollow tells the file tailer to open with O_NOFOLLOW on each component.
+			cfg.NoFollow = true
+		}
 
 		if service != nil {
 			// a config defined in a container label or a pod annotation does not always contain a type,

pkg/logs/schedulers/ad/scheduler_test.go

@@ -496,3 +496,59 @@ func TestCreateSourcesWithNilConfigurations(t *testing.T) {
 		})
 	}
 }
+
+// TestNoFollowFlagSetOnConfig verifies that sources created from a process_log
+// provider have NoFollow=true, and that sources from other providers do not.
+// This flag is used by the file tailer to enforce O_NOFOLLOW opens, preventing
+// symlink-swap attacks on process_log-discovered paths.
+func TestNoFollowFlagSetOnConfig(t *testing.T) {
+	tests := []struct {
+		name         string
+		provider     string
+		expectedFlag bool
+	}{
+		{
+			name:         "process_log provider sets NoFollow=true",
+			provider:     names.ProcessLog,
+			expectedFlag: true,
+		},
+		{
+			name:         "file provider does not set NoFollow",
+			provider:     names.File,
+			expectedFlag: false,
+		},
+		{
+			name:         "kubernetes provider does not set NoFollow",
+			provider:     names.Kubernetes,
+			expectedFlag: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var logsConfig []byte
+			var serviceID string
+
+			if tt.provider == names.File {
+				logsConfig = []byte("logs:\n  - type: file\n    path: /var/log/app.log\n    service: svc\n")
+				serviceID = ""
+			} else {
+				logsConfig = []byte(`[{"type":"file","path":"/var/log/app.log","service":"svc"}]`)
+				serviceID = tt.provider + ":///var/log/app.log"
+			}
+
+			cfg := integration.Config{
+				LogsConfig: logsConfig,
+				Provider:   tt.provider,
+				Name:       "test-config",
+				ServiceID:  serviceID,
+			}
+
+			sources, err := CreateSources(cfg)
+			require.NoError(t, err)
+			require.Len(t, sources, 1)
+			assert.Equal(t, tt.expectedFlag, sources[0].Config.NoFollow,
+				"Config.NoFollow should be %v for provider %q", tt.expectedFlag, tt.provider)
+		})
+	}
+}