[Ingest Management] Fixed parsing of npipe URI

libbeat/api/npipe/listener_windows.go

@@ -98,5 +98,14 @@ func DefaultSD(forUser string) (string, error) {
 	// String definition: https://docs.microsoft.com/en-us/windows/win32/secauthz/ace-strings
 	// Give generic read/write access to the specified user.
 	descriptor := "D:P(A;;GA;;;" + u.Uid + ")"
+	if u.Username == "NT AUTHORITY\\SYSTEM" {
+		// running as SYSTEM, include Administrators group so Administrators can talk over
+		// the named pipe to the running Elastic Agent system process
+		admin, err := user.LookupGroup("Administrators")
+		if err != nil {
+			return "", errors.Wrap(err, "failed to lookup Administrators group")
+		}
+		descriptor += "(A;;GA;;;" + admin.Gid + ")"
+	}
 	return descriptor, nil
 }

metricbeat/mb/parse/url.go

@@ -212,12 +212,29 @@ func getURL(
 		u.Scheme = "http"
 		u.Host = "unix"
 	case "http+npipe":
-		p := strings.Replace(u.Path, "/pipe", `\\.\pipe`, 1)
-		p = strings.Replace(p, "/", "\\", -1)
-		t = dialer.NewNpipeDialerBuilder(p)
+		p := u.Path
 		u.Path = ""
 		u.Scheme = "http"
 		u.Host = "npipe"
+
+		if p == "" && u.Host != "" {
+			p = u.Host
+		}
+
+		// cleanup of all possible prefixes
+		p = strings.TrimPrefix(p, "/pipe")
+		p = strings.TrimPrefix(p, `\\.\pipe`)
+		p = strings.TrimPrefix(p, "\\")
+		p = strings.TrimPrefix(p, "/")
+
+		segs := strings.SplitAfterN(p, "/", 2)
+		if len(segs) == 2 {
+			p = strings.TrimSuffix(segs[0], "/")
+			u.Path = "/" + segs[1]
+		}
+
+		p = `\\.\pipe\` + strings.Replace(p, "/", "\\", -1)
+		t = dialer.NewNpipeDialerBuilder(p)
 	default:
 		t = dialer.NewDefaultDialerBuilder()
 	}

metricbeat/mb/parse/url_test.go

@@ -121,6 +121,21 @@ func TestParseURL(t *testing.T) {
 		}
 	})
 
+	t.Run("http+npipe short with", func(t *testing.T) {
+		rawURL := "http+npipe:///custom"
+		hostData, err := ParseURL(rawURL, "http", "", "", "apath", "")
+		if assert.NoError(t, err) {
+			transport, ok := hostData.Transport.(*dialer.NpipeDialerBuilder)
+			assert.True(t, ok)
+			assert.Equal(t, `\\.\pipe\custom`, transport.Path)
+			assert.Equal(t, "http://npipe/apath", hostData.URI)
+			assert.Equal(t, "http://npipe/apath", hostData.SanitizedURI)
+			assert.Equal(t, "npipe", hostData.Host)
+			assert.Equal(t, "", hostData.User)
+			assert.Equal(t, "", hostData.Password)
+		}
+	})
+
 	t.Run("npipe", func(t *testing.T) {
 		rawURL := "npipe://./pipe/docker_engine"
 		hostData, err := ParseURL(rawURL, "tcp", "", "", "", "")

x-pack/elastic-agent/CHANGELOG.next.asciidoc

@@ -27,6 +27,7 @@
 - Fix deb/rpm packaging for Elastic Agent {pull}22153[22153]
 - Fix composable input processor promotion to fix duplicates {pull}22344[22344]
 - Fix sysv init files for deb/rpm installation {pull}22543[22543]
+- Fixed parsing of npipe URI {pull}22978[22978]
 
 ==== New features