AVRO-4314: [csharp] Validate names against the Avro name grammar

lang/csharp/src/apache/main/Protocol/Message.cs

@@ -76,6 +76,7 @@ public class Message
         public Message(string name, string doc, RecordSchema request, Schema response, UnionSchema error, bool? oneway)
         {
             if (string.IsNullOrEmpty(name)) throw new ArgumentNullException(nameof(name), "name cannot be null.");
+            SchemaName.ValidateName(name, "message");
             this.Request = request;
             this.Response = response;
             this.Error = error;

lang/csharp/src/apache/main/Schema/Field.cs

@@ -153,6 +153,16 @@ internal Field(Schema schema, string name, IList<string> aliases, int pos, strin
                 throw new ArgumentNullException(nameof(name), "name cannot be null.");
             }
 
+            SchemaName.ValidateName(name, "field");
+
+            if (aliases != null)
+            {
+                foreach (string alias in aliases)
+                {
+                    SchemaName.ValidateName(alias, "field alias");
+                }
+            }
+
             Schema = schema ?? throw new ArgumentNullException("type", "type cannot be null.");
             Name = name;
             Aliases = aliases;

lang/csharp/src/apache/main/Schema/SchemaName.cs

@@ -17,6 +17,8 @@
  */
 using System;
 using System.Collections.Generic;
+using System.Globalization;
+using System.Text;
 
 namespace Avro
 {
@@ -25,6 +27,83 @@ namespace Avro
     /// </summary>
     public class SchemaName
     {
+        /// <summary>
+        /// Validates that a simple (unqualified) name conforms to the Avro name
+        /// grammar: a non-empty string whose first character is a letter or '_'
+        /// and whose remaining characters are letters, digits or '_'. Letters and
+        /// digits are recognized in a Unicode-aware way (including supplementary
+        /// characters represented by surrogate pairs), matching the default
+        /// behavior of the Java SDK (and the C# SDK's existing support for
+        /// non-ASCII field names). Throws <see cref="SchemaParseException"/> when
+        /// the name is invalid.
+        /// </summary>
+        /// <param name="name">the simple name to validate</param>
+        /// <param name="what">description of the kind of name, used in error messages</param>
+        internal static void ValidateName(string name, string what)
+        {
+            if (string.IsNullOrEmpty(name)
+                || !(name[0] == '_' || char.IsLetter(name, 0)))
+            {
+                throw new SchemaParseException($"Invalid {what} name: {Quote(name)}");
+            }
+
+            // Iterate by Unicode scalar value so supplementary-plane letters and
+            // digits (encoded as surrogate pairs) are handled correctly.
+            int i = char.IsSurrogatePair(name, 0) ? 2 : 1;
+            while (i < name.Length)
+            {
+                if (name[i] == '_' || char.IsLetterOrDigit(name, i))
+                {
+                    i += char.IsSurrogatePair(name, i) ? 2 : 1;
+                }
+                else
+                {
+                    throw new SchemaParseException($"Invalid {what} name: {Quote(name)}");
+                }
+            }
+        }
+
+        /// <summary>
+        /// Returns a quoted representation of the given value with control
+        /// characters escaped, so that an invalid name embedded in an error
+        /// message cannot inject newlines or other control characters.
+        /// </summary>
+        private static string Quote(string value)
+        {
+            if (value == null)
+            {
+                return "null";
+            }
+
+            var sb = new StringBuilder(value.Length + 2);
+            sb.Append('"');
+            foreach (char c in value)
+            {
+                switch (c)
+                {
+                    case '"': sb.Append("\\\""); break;
+                    case '\\': sb.Append("\\\\"); break;
+                    case '\r': sb.Append("\\r"); break;
+                    case '\n': sb.Append("\\n"); break;
+                    case '\t': sb.Append("\\t"); break;
+                    default:
+                        if (char.IsControl(c))
+                        {
+                            sb.AppendFormat(CultureInfo.InvariantCulture, "\\u{0:x4}", (int)c);
+                        }
+                        else
+                        {
+                            sb.Append(c);
+                        }
+
+                        break;
+                }
+            }
+
+            sb.Append('"');
+            return sb.ToString();
+        }
+
         // cache the full name, so it won't allocate new strings on each call
         private String fullName;
 
@@ -88,6 +167,16 @@ public SchemaName(string name, string space, string encspace, string documentati
 
             Documentation = documentation;
             fullName = string.IsNullOrEmpty(Namespace) ? Name : Namespace + "." + Name;
+
+            // Validate the simple name only. Namespaces are intentionally not
+            // validated here: the code generator's namespace-mapping feature
+            // rewrites schema namespaces to C#-specific values (for example
+            // "@return" for reserved words) and re-parses the schema, so a
+            // namespace component may legitimately not match the Avro name grammar.
+            if (Name != null)
+            {
+                ValidateName(Name, "schema");
+            }
         }
 
         /// <summary>

lang/csharp/src/apache/test/Protocol/ProtocolTest.cs

@@ -181,6 +181,23 @@ public static void TestProtocol(string str, bool valid)
             Assert.AreEqual(json,json2);
         }
 
+        [TestCase(@"{
+  ""protocol"": ""P"",
+  ""namespace"": ""com.acme"",
+  ""types"": [],
+  ""messages"": { ""bad name"": { ""request"": [], ""response"": ""null"" } }
+}", TestName = "MessageNameWithSpace")]
+        [TestCase(@"{
+  ""protocol"": ""P"",
+  ""namespace"": ""com.acme"",
+  ""types"": [],
+  ""messages"": { ""x; int y"": { ""request"": [], ""response"": ""null"" } }
+}", TestName = "MessageNameInjection")]
+        public static void TestInvalidMessageName(string str)
+        {
+            Assert.Throws<ProtocolParseException>(() => Protocol.Parse(str));
+        }
+
         // Protocols match
         [TestCase(
 @"{

lang/csharp/src/apache/test/Schema/SchemaTests.cs

@@ -109,6 +109,16 @@ public class SchemaTests
         [TestCase("{\"type\": \"fixed\", \"name\": \"Missing size\"}", typeof(SchemaParseException))]
         [TestCase("{\"type\": \"fixed\", \"size\": 314}",
             typeof(SchemaParseException), Description = "No name")]
+
+        // Names outside the Avro name grammar
+        [TestCase("{\"type\":\"record\",\"name\":\"Bad Name\",\"fields\":[]}",
+            typeof(SchemaParseException), Description = "Record name with a space")]
+        [TestCase("{\"type\":\"record\",\"name\":\"R\",\"fields\":[{\"name\":\"in valid\",\"type\":\"long\"}]}",
+            typeof(SchemaParseException), Description = "Field name with a space")]
+        [TestCase("{\"type\":\"record\",\"name\":\"R\",\"fields\":[{\"name\":\"x; int y\",\"type\":\"long\"}]}",
+            typeof(SchemaParseException), Description = "Field name attempting identifier injection")]
+        [TestCase("{\"type\":\"record\",\"name\":\"R\",\"fields\":[{\"name\":\"valid\",\"aliases\":[\"bad alias\"],\"type\":\"long\"}]}",
+            typeof(SchemaParseException), Description = "Field alias with a space")]
         public void TestBasic(string s, Type expectedExceptionType = null)
         {
             if (expectedExceptionType != null)
@@ -378,6 +388,17 @@ public void TestRecordFieldNames() {
 
             Field f = recordSchema.Fields[0];
             Assert.AreEqual("歳以上", f.Name);
+
+            // A supplementary-plane letter (U+20000, encoded as a surrogate pair)
+            // is a valid name character and must be accepted.
+            const string astralName = "\U00020000field";
+            var astralFields = new List<Field>
+                {
+                    new Field(PrimitiveSchema.Create(Schema.Type.Long), astralName, null, 0, null, null,
+                        Field.SortOrder.ignore, null)
+                };
+            var astralRecord = RecordSchema.Create("AstralRecord", astralFields);
+            Assert.AreEqual(astralName, astralRecord.Fields[0].Name);
         }
 
         [TestCase]