Add Avro crypto shredding

.editorconfig

@@ -137,4 +137,7 @@ indent_size = 2
 indent_size = 2
 
 [*.csproj]
-indent_size = 2
+indent_size = 2
+
+[*.avsc]
+indent_size = 2

KafkaFlow.Contrib.sln

@@ -23,6 +23,12 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Contrib.KafkaFlow.ProcessMa
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Contrib.KafkaFlow.SqlServer", "src\Contrib.KafkaFlow.SqlServer\Contrib.KafkaFlow.SqlServer.csproj", "{6839FA17-DF28-418B-AB5A-2249D44F4CA3}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "KafkaFlow.CryptoShredding.Tests", "tests\KafkaFlow.CryptoShredding.Tests\KafkaFlow.CryptoShredding.Tests.csproj", "{3C8FF880-DED3-4D78-A6CD-47FADCDE2A22}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Contrib.KafkaFlow.CryptoShredding.Avro", "src\Contrib.KafkaFlow.CryptoShredding.Avro\Contrib.KafkaFlow.CryptoShredding.Avro.csproj", "{9916233A-864C-4EB6-B623-46183E165705}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Contrib.KafkaFlow.CryptoShredding.Avro.Analyzers", "src\Contrib.KafkaFlow.CryptoShredding.Avro.Analyzers\Contrib.KafkaFlow.CryptoShredding.Avro.Analyzers.csproj", "{813CCAC0-9FBD-4A1A-B063-A4F1F9B42688}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -61,6 +67,18 @@ Global
 		{6839FA17-DF28-418B-AB5A-2249D44F4CA3}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{6839FA17-DF28-418B-AB5A-2249D44F4CA3}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{6839FA17-DF28-418B-AB5A-2249D44F4CA3}.Release|Any CPU.Build.0 = Release|Any CPU
+		{3C8FF880-DED3-4D78-A6CD-47FADCDE2A22}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{3C8FF880-DED3-4D78-A6CD-47FADCDE2A22}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{3C8FF880-DED3-4D78-A6CD-47FADCDE2A22}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{3C8FF880-DED3-4D78-A6CD-47FADCDE2A22}.Release|Any CPU.Build.0 = Release|Any CPU
+		{9916233A-864C-4EB6-B623-46183E165705}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{9916233A-864C-4EB6-B623-46183E165705}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{9916233A-864C-4EB6-B623-46183E165705}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{9916233A-864C-4EB6-B623-46183E165705}.Release|Any CPU.Build.0 = Release|Any CPU
+		{813CCAC0-9FBD-4A1A-B063-A4F1F9B42688}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{813CCAC0-9FBD-4A1A-B063-A4F1F9B42688}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{813CCAC0-9FBD-4A1A-B063-A4F1F9B42688}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{813CCAC0-9FBD-4A1A-B063-A4F1F9B42688}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -74,6 +92,9 @@ Global
 		{A4CBB313-9DCC-441E-BB44-1473CDF7A986} = {CE99781E-E88C-404C-B9AD-D7F713265F31}
 		{3F0C0AD0-CFD8-4722-BF3B-6259D0878680} = {CE99781E-E88C-404C-B9AD-D7F713265F31}
 		{6839FA17-DF28-418B-AB5A-2249D44F4CA3} = {CE99781E-E88C-404C-B9AD-D7F713265F31}
+		{3C8FF880-DED3-4D78-A6CD-47FADCDE2A22} = {A5E2EC13-2E8B-4EB3-8C0F-D252A04B42F4}
+		{9916233A-864C-4EB6-B623-46183E165705} = {CE99781E-E88C-404C-B9AD-D7F713265F31}
+		{813CCAC0-9FBD-4A1A-B063-A4F1F9B42688} = {CE99781E-E88C-404C-B9AD-D7F713265F31}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {66EE15D8-69C4-4AFB-9EB8-6020B2D8C01A}

src/Contrib.KafkaFlow.CryptoShredding.Avro.Analyzers/AvroSourceGenerator.cs

@@ -0,0 +1,72 @@
+using System.CodeDom;
+using System.CodeDom.Compiler;
+using System.Text;
+using Avro;
+using Avro.Util;
+using KafkaFlow.CryptoShredding.Avro;
+using Avro.LogicalTypes;
+using Microsoft.CodeAnalysis;
+using Microsoft.CodeAnalysis.Text;
+using Microsoft.CSharp;
+
+namespace KafkaFlow.CryptoShredding.Avro.Analyzers;
+
+[Generator]
+public class AvroSourceGenerator : ISourceGenerator
+{
+    private static readonly CodeGeneratorOptions CodeGenOptions =
+        new() { BracingStyle = "C", BlankLinesBetweenMembers = false };
+
+    public void Initialize(GeneratorInitializationContext context)
+    {
+        LogicalTypeFactory.Instance.Register(new InlineEncryptedStringLogicalType());
+        LogicalTypeFactory.Instance.Register(new EncryptedStringLogicalType());
+    }
+
+    public void Execute(GeneratorExecutionContext context)
+    {
+        var codeGen = new CodeGen();
+
+        var schemas = context.AdditionalFiles
+            .Where(x => Path.GetExtension(x.Path).Equals(".avsc", StringComparison.OrdinalIgnoreCase))
+            .Select(x => x.GetText())
+            .Select(x => x is null ? null : Schema.Parse(x.ToString()))
+            .Where(x => x is not null);
+
+        foreach (var schema in schemas) codeGen.AddSchema(schema);
+
+        GenerateCodeForTypes(context, codeGen.GenerateCode());
+    }
+
+    // There is a bug in .NET avrogen 'GetTypes': it doesn't correctly handle types with the same
+    // name but in different namespaces.
+    // This method uses namespaces as folders to avoid this issue.
+    private static void GenerateCodeForTypes(GeneratorExecutionContext context, CodeCompileUnit code)
+    {
+        using var provider = new CSharpCodeProvider();
+
+        for (var i = 0; i < code.Namespaces.Count; i++)
+        {
+            var ns = code.Namespaces[i];
+            var path = CodeGenUtil.Instance.UnMangle(ns.Name).Split('.').Aggregate(Path.Combine);
+
+            var newNs = new CodeNamespace(ns.Name);
+            newNs.Comments.Add(CodeGenUtil.Instance.FileComment);
+            newNs.Imports.AddRange(CodeGenUtil.Instance.NamespaceImports);
+
+            for (var j = 0; j < ns.Types.Count; j++)
+            {
+                var type = ns.Types[j];
+                newNs.Types.Add(type);
+                using (var writer = new StringWriter())
+                {
+                    var fileName = Path.ChangeExtension(CodeGenUtil.Instance.UnMangle(type.Name), "cs");
+                    provider.GenerateCodeFromNamespace(newNs, writer, CodeGenOptions);
+                    context.AddSource(Path.Combine(path, fileName), SourceText.From(writer.ToString(), Encoding.UTF8));
+                }
+
+                newNs.Types.Clear();
+            }
+        }
+    }
+}

src/Contrib.KafkaFlow.CryptoShredding.Avro.Analyzers/Contrib.KafkaFlow.CryptoShredding.Avro.Analyzers.csproj

@@ -0,0 +1,52 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFrameworks>netstandard2.0;netstandard2.1</TargetFrameworks>
+    <LangVersion>12</LangVersion>
+    <RootNamespace>KafkaFlow.CryptoShredding.Avro.Analyzers</RootNamespace>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+    <EnforceExtendedAnalyzerRules>true</EnforceExtendedAnalyzerRules>
+    <IsRoslynComponent>true</IsRoslynComponent>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Apache.Avro" Version="1.11.3" GeneratePathProperty="true" PrivateAssets="all"/>
+    <PackageReference Include="Microsoft.CodeAnalysis.Common" Version="4.10.0"/>
+    <PackageReference Include="Newtonsoft.Json" Version="13.0.3" GeneratePathProperty="true" PrivateAssets="all"/>
+    <PackageReference Include="System.CodeDom" Version="8.0.0" GeneratePathProperty="true" PrivateAssets="all"/>
+    <PackageReference Include="IsExternalInit" Version="1.0.3" PrivateAssets="all"/>
+    <PackageReference Include="Nullable" Version="1.3.1" PrivateAssets="all"/>
+  </ItemGroup>
+
+  <PropertyGroup>
+    <GetTargetPathDependsOn>$(GetTargetPathDependsOn);GetDependencyTargetPaths</GetTargetPathDependsOn>
+  </PropertyGroup>
+
+  <Target Name="GetDependencyTargetPaths">
+    <ItemGroup>
+      <TargetPathWithTargetPlatformMoniker Include="$(PkgApache_Avro)\lib\netstandard2.0\Avro.dll"
+                                           IncludeRuntimeDependency="false"/>
+      <TargetPathWithTargetPlatformMoniker
+        Include="$(PkgNewtonsoft_Json)\lib\netstandard2.0\Newtonsoft.Json.dll"
+        IncludeRuntimeDependency="false"/>
+      <TargetPathWithTargetPlatformMoniker
+        Include="$(PkgSystem_CodeDom)\lib\netstandard2.0\System.CodeDom.dll"
+        IncludeRuntimeDependency="false"/>
+      <!-- Pack both our DLL and the dependencies into a generated Nuget Package -->
+      <None Include="$(OutputPath)\$(AssemblyName).dll" Pack="true"
+            PackagePath="analyzers/dotnet/cs" Visible="false"/>
+      <None Include="$(PkgApache_Avro)\lib\netstandard2.0\Avro.dll" Pack="true"
+            PackagePath="analyzers/dotnet/cs" Visible="false"/>
+      <None Include="$(PkgNewtonsoft_Json)\lib\netstandard2.0\Newtonsoft.Json.dll" Pack="true"
+            PackagePath="analyzers/dotnet/cs" Visible="false"/>
+      <None Include="$(PkgSystem_CodeDom)\lib\netstandard2.0\System.CodeDom.dll" Pack="true"
+            PackagePath="analyzers/dotnet/cs" Visible="false"/>
+    </ItemGroup>
+  </Target>
+
+  <ItemGroup>
+    <Compile Include="..\Contrib.KafkaFlow.CryptoShredding.Avro\LogicalTypes\*.cs"/>
+  </ItemGroup>
+
+</Project>

src/Contrib.KafkaFlow.CryptoShredding.Avro/AesKeyedEncryptor.cs

@@ -0,0 +1,54 @@
+using System.Security.Cryptography;
+using System.Text;
+
+namespace KafkaFlow.CryptoShredding.Avro;
+
+public interface IKeyedEncryptor : IDisposable
+{
+    string Encrypt(string plainText);
+    string Decrypt(string cipherText);
+}
+
+public sealed class AesKeyedEncryptor : IKeyedEncryptor
+{
+    private readonly Aes _aes;
+
+    public AesKeyedEncryptor(string key)
+    {
+        _aes = Aes.Create();
+        using var sha256 = SHA256.Create();
+        _aes.Key = sha256.ComputeHash(Encoding.UTF8.GetBytes(key));
+
+        var iv = new byte[16];
+        Array.Copy(_aes.Key, iv, 16);
+        _aes.IV = iv;
+    }
+
+    public string Encrypt(string plainText)
+    {
+        using var encryptor = _aes.CreateEncryptor();
+        using var ms = new MemoryStream();
+        using (var cs = new CryptoStream(ms, encryptor, CryptoStreamMode.Write))
+        {
+            using var sw = new StreamWriter(cs);
+            sw.Write(plainText);
+        }
+        var result = Convert.ToBase64String(ms.ToArray());
+        return result;
+    }
+
+    public string Decrypt(string cipherText)
+    {
+        var buffer = Convert.FromBase64String(cipherText);
+        using var decryptor = _aes.CreateDecryptor();
+        using var ms = new MemoryStream(buffer);
+        using var cs = new CryptoStream(ms, decryptor, CryptoStreamMode.Read);
+        using var sr = new StreamReader(cs);
+        return sr.ReadToEnd();
+    }
+
+    public void Dispose()
+    {
+        _aes.Dispose();
+    }
+}