Document minimum privileges (#323)

docs/data-shippers/index.asciidoc

@@ -12,6 +12,43 @@ sending events (such as logs) to various outputs.
 Currently these shippers support Elastic Cloud & Elasticsearch but
 other outputs are in the works.
 
+=== Elasticsearch Security
+
+If Elasticsearch's security is enabled you will need to ensure you configure a user or API key with enough privileges
+
+==== Bootstrap 
+
+In order for the datashippers to have enough privileges to bootstrap the
+target datastreams with all the ECS mappings, templates and settings the authenticated 
+security principal needs the following minimum privileges:
+
+[options="header"]
+|====
+|Type | Privileges
+
+|Cluster
+|`monitor`, `manage_ilm`, `manage_index_templates`, `manage_pipeline`
+
+|Index
+|`manage`, `create_doc`
+|====
+
+==== No bootstrap
+
+If the datashippers are configured to skip bootstrapping the target destinations all together,
+the security principal requires the following minimum privileges to push data.
+
+[options="header"]
+|====
+|Type | Privileges
+
+|Cluster
+|`monitor` 
+
+|Index
+|`auto_configure` `create_doc`
+|====
+
 include::./ingest-commonschema.asciidoc[Elastic.Ingest.Elasticsearch.CommonSchema]
 include::./serilog.asciidoc[Serilog]
 include::./extensions-logging.asciidoc[Microsoft.Extensions.Logging]

src/Elastic.CommonSchema.BenchmarkDotNetExporter/ElasticsearchBenchmarkExporterOptions.cs

@@ -7,6 +7,7 @@
 using Elastic.Channels.Diagnostics;
 using Elastic.CommonSchema.BenchmarkDotNetExporter.Domain;
 using Elastic.Ingest.Elasticsearch;
+using Elastic.Ingest.Elasticsearch.CommonSchema;
 using Elastic.Ingest.Elasticsearch.DataStreams;
 using Elastic.Transport;
 using Elastic.Transport.Products.Elasticsearch;
@@ -146,7 +147,7 @@ private NodePool CreateNodePool()
 
 		internal TransportConfiguration CreateTransportConfiguration()
 		{
-			var settings = new TransportConfiguration(CreateNodePool(), productRegistration: new ElasticsearchProductRegistration());
+			var settings = new TransportConfiguration(CreateNodePool(), productRegistration: ElasticsearchProductRegistration.Default);
 			if (EnableDebugMode)
 				settings.EnableDebugMode();
 			return settings;

src/Elastic.Extensions.Logging/ElasticsearchLoggerProvider.cs

@@ -132,7 +132,7 @@ private static HttpTransport CreateTransport(ElasticsearchLoggerOptions loggerOp
 			if (loggerOptions.Transport != null) return loggerOptions.Transport;
 
 			var connectionPool = CreateNodePool(loggerOptions);
-			var config = new TransportConfiguration(connectionPool, productRegistration: new ElasticsearchProductRegistration());
+			var config = new TransportConfiguration(connectionPool, productRegistration: ElasticsearchProductRegistration.Default);
 			// Cloud sets authentication as required parameter in the constructor
 			if (loggerOptions.ShipTo.NodePoolType != NodePoolType.Cloud)
 				config = SetAuthenticationOnTransport(loggerOptions, config);

src/Elastic.Ingest.Elasticsearch.CommonSchema/Elastic.Ingest.Elasticsearch.CommonSchema.csproj

@@ -10,7 +10,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="Elastic.Ingest.Elasticsearch" Version="0.5.3" />
+    <PackageReference Include="Elastic.Ingest.Elasticsearch" Version="0.5.5" />
   </ItemGroup>
 
 </Project>

src/Elastic.Serilog.Sinks/TransportHelper.cs

@@ -9,7 +9,7 @@ namespace Elastic.Serilog.Sinks
 {
 	internal static class TransportHelper
 	{
-		private static readonly ProductRegistration DefaultProduct = new ElasticsearchProductRegistration();
+		private static readonly ProductRegistration DefaultProduct = ElasticsearchProductRegistration.Default;
 
 		public static TransportConfiguration Default() =>
 			new TransportConfiguration(new Uri("http://localhost:9200"), DefaultProduct);

tests-integration/Elastic.Serilog.Sinks.IntegrationTests/BootstrapMinimumSecurityPrivilegesTests.cs

@@ -0,0 +1,143 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Elastic.Channels;
+using Elastic.Channels.Diagnostics;
+using Elastic.Clients.Elasticsearch;
+using Elastic.Clients.Elasticsearch.IndexManagement;
+using Elastic.CommonSchema;
+using Elastic.Elasticsearch.Xunit.XunitPlumbing;
+using Elastic.Ingest.Elasticsearch;
+using Elastic.Transport;
+using Elasticsearch.IntegrationDefaults;
+using FluentAssertions;
+using Serilog;
+using Serilog.Core;
+using Xunit.Abstractions;
+using DataStreamName = Elastic.Ingest.Elasticsearch.DataStreams.DataStreamName;
+
+namespace Elastic.Serilog.Sinks.IntegrationTests;
+
+public class BootstrapMinimumSecurityPrivilegesTests : SecurityPrivilegesTestsBase
+{
+	public BootstrapMinimumSecurityPrivilegesTests(SecurityCluster cluster, ITestOutputHelper output)
+		: base(cluster, output)
+	{
+	}
+
+	protected override BootstrapMethod Bootstrap => BootstrapMethod.Failure;
+	protected override DataStreamName Target { get; } = new ("logs", "serilog.setup");
+
+	protected override string ApiKeyJson => $@"{{
+	""name"": ""ecs_setup"",
+	""role_descriptors"": {{
+		""ecs_setup"": {{
+			""cluster"": [""monitor"", ""manage_ilm"", ""manage_index_templates"", ""manage_pipeline""],
+			""index"": [{{
+				""names"": [""{Target.GetNamespaceWildcard()}""],
+				""privileges"": [""manage"", ""create_doc""]
+			}}]
+		}}
+	}}
+}}";
+}
+
+public class NoBootstrapMinimumSecurityPrivilegesTests : SecurityPrivilegesTestsBase
+{
+	public NoBootstrapMinimumSecurityPrivilegesTests(SecurityCluster cluster, ITestOutputHelper output)
+		: base(cluster, output)
+	{
+	}
+
+	protected override BootstrapMethod Bootstrap => BootstrapMethod.None;
+	protected override DataStreamName Target { get; } = new ("logs", "serilog.write");
+
+	protected override string ApiKeyJson => $@"{{
+	""name"": ""ecs_write"",
+	""role_descriptors"": {{
+		""ecs_write"": {{
+			""cluster"": [""monitor""],
+			""index"": [{{
+				""names"": [""{Target.GetNamespaceWildcard()}""],
+				""privileges"": [""auto_configure"", ""create_doc""]
+			}}]
+		}}
+	}}
+}}";
+}
+
+public abstract class SecurityPrivilegesTestsBase : SerilogTestBase<SecurityCluster>
+{
+	private IChannelDiagnosticsListener? _listener;
+	private readonly CountdownEvent _waitHandle = new(1);
+	private ElasticsearchSinkOptions SinkOptions { get; }
+
+	private ElasticsearchClient ApiScopedClient { get; }
+
+	protected abstract string ApiKeyJson { get; }
+	protected abstract DataStreamName Target { get; }
+	protected abstract BootstrapMethod Bootstrap { get; }
+
+	protected SecurityPrivilegesTestsBase(SecurityCluster cluster, ITestOutputHelper output) : base(cluster, output)
+	{
+		var logs = new List<Action<Logger>>
+		{
+			l => l.Information("Hello Information"),
+			l => l.Debug("Hello Debug"),
+			l => l.Warning("Hello Warning"),
+			l => l.Error("Hello Error"),
+			l => l.Fatal("Hello Fatal")
+		};
+
+		var apiKey = cluster.CreateApiKey(Client, ApiKeyJson);
+
+		ApiScopedClient = cluster.CreateElasticsearchClient(output,
+			s=>s.Authentication(new ApiKey(apiKey.Encoded))
+		);
+
+		SinkOptions = new ElasticsearchSinkOptions(ApiScopedClient.Transport)
+		{
+			BootstrapMethod = Bootstrap,
+			DataStream = Target,
+			ConfigureChannel = c =>
+			{
+				c.BufferOptions = new BufferOptions
+				{
+					WaitHandle = _waitHandle,
+					OutboundBufferMaxSize = logs.Count
+				};
+			},
+			ChannelDiagnosticsCallback = (l) => _listener = l
+		};
+
+		var loggerConfig = new LoggerConfiguration()
+			.MinimumLevel.Information()
+			.WriteTo.ColoredConsole()
+			.WriteTo.Elasticsearch(SinkOptions);
+
+		using var logger = loggerConfig.CreateLogger();
+		foreach (var a in logs) a(logger);
+	}
+
+	[I] public async Task AssertLogs()
+	{
+		if (!_waitHandle.WaitHandle.WaitOne(TimeSpan.FromSeconds(10)))
+			throw new Exception($"No flush occurred in 10 seconds: {_listener}", _listener?.ObservedException);
+
+		var indexName = SinkOptions.DataStream.ToString();
+		var refreshed = await Client.Indices.RefreshAsync(new RefreshRequest(indexName));
+		refreshed.IsValidResponse.Should().BeTrue("{0}", refreshed.DebugInformation);
+
+		var search = await Client.SearchAsync<EcsDocument>(new SearchRequest(indexName));
+
+		// Informational should be filtered
+		search.Documents.Count().Should().Be(4);
+
+		var messages = search.Documents.Select(e => e.Message);
+		messages.Should().Contain("Hello Error");
+	}
+
+
+}

tests-integration/Elastic.Serilog.Sinks.IntegrationTests/SerilogCluster.cs

@@ -1,7 +1,13 @@
+using System.Text.Json.Serialization;
+using Elastic.Clients.Elasticsearch;
+using Elastic.CommonSchema;
 using Elastic.Elasticsearch.Ephemeral;
 using Elastic.Elasticsearch.Xunit;
+using Elastic.Transport;
+using Elastic.Transport.Products.Elasticsearch;
 using Elasticsearch.IntegrationDefaults;
 using Xunit;
+using static Elastic.Elasticsearch.Ephemeral.ClusterAuthentication;
 
 [assembly: TestFramework("Elastic.Elasticsearch.Xunit.Sdk.ElasticTestFramework", "Elastic.Elasticsearch.Xunit")]
 
@@ -19,4 +25,29 @@ public SecurityCluster() : base(9206, ClusterFeatures.XPack | ClusterFeatures.Se
 	{
 
 	}
+
+	protected override ElasticsearchClientSettings UpdateClientSettings(ElasticsearchClientSettings settings) =>
+		settings.Authentication(new BasicAuthentication(Admin.Username, Admin.Password));
+
+	public ApiKeyResponse CreateApiKey(ElasticsearchClient client, string json)
+	{
+
+		var apiKey =  client.Transport.Request<ApiKeyResponse>(HttpMethod.POST, "/_security/api_key", PostData.String(json));
+		return apiKey;
+	}
+
+	public class ApiKeyResponse : ElasticsearchResponse
+	{
+		[JsonPropertyName("id")]
+		public string Id { get; init; } = default!;
+
+		[JsonPropertyName("name")]
+		public string Name { get; init; } = default!;
+
+		[JsonPropertyName("api_key")]
+		public string ApiKey { get; init; } = default!;
+
+		[JsonPropertyName("encoded")]
+		public string Encoded { get; init; } = default!;
+	}
 }

tests-integration/Elasticsearch.IntegrationDefaults/Elasticsearch.IntegrationDefaults.csproj

@@ -11,7 +11,7 @@
   <ItemGroup>
     <PackageReference Include="Elastic.Clients.Elasticsearch" Version="8.0.4" />
     <PackageReference Include="Elastic.Elasticsearch.Xunit" Version="0.4.3" />
-    <PackageReference Include="Elastic.Ingest.Elasticsearch" Version="0.5.3" />
+    <PackageReference Include="Elastic.Ingest.Elasticsearch" Version="0.5.5" />
 
   </ItemGroup>
 

tests-integration/Elasticsearch.IntegrationDefaults/IngestionCluster.cs

@@ -13,52 +13,61 @@
 
 [assembly: TestFramework("Elastic.Elasticsearch.Xunit.Sdk.ElasticTestFramework", "Elastic.Elasticsearch.Xunit")]
 
-namespace Elasticsearch.IntegrationDefaults
+namespace Elasticsearch.IntegrationDefaults;
+
+public static class TestClusterExtensions
 {
-	/// <summary> Declare our cluster that we want to inject into our test classes </summary>
-	public abstract class TestClusterBase : XunitClusterBase
+	public static ElasticsearchClient CreateElasticsearchClient(
+		this IEphemeralCluster cluster,
+		ITestOutputHelper output,
+		Func<ElasticsearchClientSettings, ElasticsearchClientSettings> updateSettings,
+		Func<ICollection<Uri>, ICollection<Uri>>? alterNodes = null
+	)
 	{
-		protected TestClusterBase(int port = 9200, ClusterFeatures features = ClusterFeatures.None)
-			: base(new XunitClusterConfiguration("8.4.0", features) { StartingPortNumber = port, AutoWireKnownProxies = true }) { }
-
-		public ElasticsearchClient CreateClient(ITestOutputHelper output, Func<ICollection<Uri>, ICollection<Uri>>? alterNodes = null) =>
-			this.GetOrAddClient(cluster =>
+		var isCi = !string.IsNullOrEmpty(Environment.GetEnvironmentVariable("CI"));
+		var nodes = cluster.NodesUris();
+		if (alterNodes != null) nodes = alterNodes(nodes);
+		var connectionPool = new StaticNodePool(nodes);
+		var settings = new ElasticsearchClientSettings(connectionPool)
+			.RequestTimeout(TimeSpan.FromSeconds(5))
+			.ServerCertificateValidationCallback(CertificateValidations.AllowAll)
+			.OnRequestCompleted(d =>
 			{
-				var isCi = !string.IsNullOrEmpty(Environment.GetEnvironmentVariable("CI"));
-				var nodes = NodesUris();
-				if (alterNodes != null) nodes = alterNodes(nodes);
-				var connectionPool = new StaticNodePool(nodes);
-				var settings = new ElasticsearchClientSettings(connectionPool)
-					.RequestTimeout(TimeSpan.FromSeconds(5))
-					.ServerCertificateValidationCallback(CertificateValidations.AllowAll)
-					.OnRequestCompleted(d =>
+				try
+				{
+					// ON CI only logged failed requests
+					// Locally we just log everything for ease of development
+					if (isCi)
 					{
-						try
-						{
-							// ON CI only logged failed requests
-							// Locally we just log everything for ease of development
-							if (isCi)
-							{
-								if (!d.HasSuccessfulStatusCode)
-									output.WriteLine(d.DebugInformation);
-							}
-							else output.WriteLine(d.DebugInformation);
-						}
-						catch
-						{
-							// ignored
-						}
-					})
-					.EnableDebugMode()
-					//do not request server stack traces on CI, too noisy
-					.IncludeServerStackTraceOnError(!isCi);
-				if (cluster.DetectedProxy != None)
+						if (!d.HasSuccessfulStatusCode)
+							output.WriteLine(d.DebugInformation);
+					}
+					else output.WriteLine(d.DebugInformation);
+				}
+				catch
 				{
-					var proxyUrl = cluster.DetectedProxy == Fiddler ? "ipv4.fiddler" : "localhost";
-					settings = settings.Proxy(new Uri($"http://{proxyUrl}:8080"), null!, null!);
+					// ignored
 				}
+			})
+			.EnableDebugMode();
+		if (cluster.DetectedProxy != None)
+		{
+			var proxyUrl = cluster.DetectedProxy == Fiddler ? "ipv4.fiddler" : "localhost";
+			settings = settings.Proxy(new Uri($"http://{proxyUrl}:8080"), null!, null!);
+		}
 
-				return new ElasticsearchClient(settings);
-			});
+		return new ElasticsearchClient(updateSettings(settings));
 	}
 }
+
+/// <summary> Declare our cluster that we want to inject into our test classes </summary>
+public abstract class TestClusterBase : XunitClusterBase
+{
+	protected TestClusterBase(int port = 9200, ClusterFeatures features = ClusterFeatures.None)
+		: base(new XunitClusterConfiguration("8.4.0", features) { StartingPortNumber = port, AutoWireKnownProxies = true }) { }
+
+	protected virtual ElasticsearchClientSettings UpdateClientSettings(ElasticsearchClientSettings settings) => settings;
+
+	public ElasticsearchClient CreateClient(ITestOutputHelper output, Func<ICollection<Uri>, ICollection<Uri>>? alterNodes = null) =>
+		this.GetOrAddClient(cluster => cluster.CreateElasticsearchClient(output, UpdateClientSettings, alterNodes));
+}