Sanitize and parse logged TLDs using JSON arrays (#3164)

In FlowReporter, we extract TLDs from EPP domain commands and log them under 'tld' and 'tlds' metadata fields to generate ICANN activity reports. Previously, invalid or extremely long TLD names (such as email addresses or long domain labels in non-validated XML payloads) could break downstream log parsing.

To prevent this issue, this change does the following:

1. Java Sanitization:
   Introduces `toLogSafeLabel(...)` in `DomainFlowUtils`, which converts ASCII to lowercase, replaces any character outside of `[a-z0-9.-]` with `-`, and limits the length to 63 characters (appending '...' if truncated). FlowReporter now uses this method on guessed TLDs before logging them.

2. SQL Modernization:
   Upgrades `epp_metrics.sql` and `epp_metrics_test.sql` to use BigQuery's native `JSON_EXTRACT_STRING_ARRAY(json, '$.tlds')` instead of the legacy `SPLIT(REGEXP_EXTRACT(JSON_EXTRACT(...)))` workaround. Because the native function extracts clean string elements, it removes the need to strip quotation marks with additional regexes and prevents parsing failures on commas, spaces, or nested structures.

SQL change tested in BigQuery.

BUG=http://b/535230985

core/src/main/java/google/registry/flows/FlowReporter.java

@@ -15,9 +15,9 @@
 package google.registry.flows;
 
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
+import static google.registry.flows.domain.DomainFlowUtils.toLogSafeLabel;
 import static java.util.Collections.EMPTY_LIST;
 
-import com.google.common.base.Ascii;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
@@ -90,9 +90,7 @@ public class FlowReporter {
    */
   private static Optional<String> extractTld(String domainName) {
     int index = domainName.indexOf('.');
-    return index == -1
-        ? Optional.empty()
-        : Optional.of(Ascii.toLowerCase(domainName.substring(index + 1)));
+    return index == -1 ? Optional.empty() : toLogSafeLabel(domainName.substring(index + 1));
   }
 
   /**

core/src/main/java/google/registry/flows/domain/DomainFlowUtils.java

@@ -14,6 +14,7 @@
 
 package google.registry.flows.domain;
 
+import static com.google.common.base.Ascii.toLowerCase;
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.base.Strings.emptyToNull;
@@ -175,6 +176,27 @@ public class DomainFlowUtils {
   /** Maximum number of characters in a domain label, from RFC 2181. */
   private static final int MAX_LABEL_SIZE = 63;
 
+  /// Given a tld name or domain label, returns a label that can be safely logged, or
+  /// `Optional.empty()` if `label` is null or blank.
+  ///
+  /// If `label` contains disallowed characters, they are replaced with '-'. If the resulting
+  /// string has a valid length, it is returned as is; if not, the first {@link #MAX_LABEL_SIZE}
+  ///  chars plus a suffix of `...` is returned.
+  ///
+  /// This method is mainly for use by `FlowReporter#recordToLogs()` to ensure that the logs
+  /// can be safely and correctly queried by SQL queries. See b/534931586 for more information.
+  public static Optional<String> toLogSafeLabel(String label) {
+    if (label == null || label.isBlank()) {
+      return Optional.empty();
+    }
+    var newLabel = ALLOWED_CHARS.negate().replaceFrom(toLowerCase(label), '-');
+    if (newLabel.length() <= MAX_LABEL_SIZE) {
+      return Optional.of(newLabel);
+    } else {
+      return Optional.of(newLabel.substring(0, MAX_LABEL_SIZE) + "...");
+    }
+  }
+
   /**
    * Returns parsed version of {@code name} if domain name label follows our naming rules and is
    * under one of the given allowed TLDs.

core/src/main/resources/google/registry/reporting/icann/sql/epp_metrics.sql

@@ -21,15 +21,12 @@
   -- "tld":"","tlds":["a.how"],"icannActivityReportField":"srs-dom-check"}
 
 SELECT
-  -- Remove quotation marks from tld fields.
-  REGEXP_EXTRACT(tld, '^"(.*)"$') AS tld,
+  tld,
   activityReportField AS metricName,
   COUNT(*) AS count
 FROM (
   SELECT
-    -- TODO(b/32486667): Replace with JSON.parse() UDF when available for views
-    SPLIT(
-    REGEXP_EXTRACT(JSON_EXTRACT(json, '$.tlds'), r'^\[(.*)\]$')) AS tlds,
+    JSON_EXTRACT_STRING_ARRAY(json, '$.tlds') AS tlds,
     JSON_EXTRACT_SCALAR(json,
       '$.resourceType') AS resourceType,
     JSON_EXTRACT_SCALAR(json,
@@ -43,10 +40,10 @@ FROM (
     WHERE
       STARTS_WITH(jsonPayload.message, "FLOW-LOG-SIGNATURE-METADATA")
       AND _TABLE_SUFFIX BETWEEN '%FIRST_DAY_OF_MONTH%' AND '%LAST_DAY_OF_MONTH%')
-  ) AS regexes
+  ) AS json_parsed
 JOIN
   -- Unnest the JSON-parsed tlds.
-  UNNEST(regexes.tlds) AS tld
+  UNNEST(json_parsed.tlds) AS tld
 -- Exclude cases that can't be tabulated correctly, where activityReportField
 -- is null/empty, or TLD is null/empty despite being a domain flow.
 WHERE

core/src/test/java/google/registry/flows/FlowReporterTest.java

@@ -184,16 +184,22 @@ class FlowReporterTest {
 
   @Test
   void testRecordToLogs_metadata_invalidDomainName_stillGuessesTld() throws Exception {
+    var maxLenTld = "a".repeat(63);
+    var domainMaxLenTld = "a." + maxLenTld;
+    var domainTldTooLong = "a." + maxLenTld + "b";
     when(flowReporter.eppInput.isDomainType()).thenReturn(true);
     when(flowReporter.eppInput.getSingleTargetId()).thenReturn(Optional.of("<foo@bar.com>"));
-    when(flowReporter.eppInput.getTargetIds()).thenReturn(ImmutableList.of("<foo@bar.com>"));
+    when(flowReporter.eppInput.getTargetIds())
+        .thenReturn(ImmutableList.of("<foo@bar.com>", domainMaxLenTld, domainTldTooLong));
     flowReporter.recordToLogs();
     Map<String, Object> json =
         parseJsonMap(findFirstLogMessageByPrefix(handler, "FLOW-LOG-SIGNATURE-METADATA: "));
     assertThat(json).containsEntry("targetId", "<foo@bar.com>");
-    assertThat(json).containsEntry("targetIds", ImmutableList.of("<foo@bar.com>"));
-    assertThat(json).containsEntry("tld", "com>");
-    assertThat(json).containsEntry("tlds", ImmutableList.of("com>"));
+    assertThat(json)
+        .containsEntry(
+            "targetIds", ImmutableList.of("<foo@bar.com>", domainMaxLenTld, domainTldTooLong));
+    assertThat(json).containsEntry("tld", "com-");
+    assertThat(json).containsEntry("tlds", ImmutableList.of("com-", maxLenTld, maxLenTld + "..."));
   }
 
   @Test

core/src/test/resources/google/registry/reporting/icann/epp_metrics_test.sql

@@ -21,15 +21,12 @@
   -- "tld":"","tlds":["a.how"],"icannActivityReportField":"srs-dom-check"}
 
 SELECT
-  -- Remove quotation marks from tld fields.
-  REGEXP_EXTRACT(tld, '^"(.*)"$') AS tld,
+  tld,
   activityReportField AS metricName,
   COUNT(*) AS count
 FROM (
   SELECT
-    -- TODO(b/32486667): Replace with JSON.parse() UDF when available for views
-    SPLIT(
-    REGEXP_EXTRACT(JSON_EXTRACT(json, '$.tlds'), r'^\[(.*)\]$')) AS tlds,
+    JSON_EXTRACT_STRING_ARRAY(json, '$.tlds') AS tlds,
     JSON_EXTRACT_SCALAR(json,
       '$.resourceType') AS resourceType,
     JSON_EXTRACT_SCALAR(json,
@@ -43,10 +40,10 @@ FROM (
     WHERE
       STARTS_WITH(jsonPayload.message, "FLOW-LOG-SIGNATURE-METADATA")
       AND _TABLE_SUFFIX BETWEEN '20170901' AND '20170930')
-  ) AS regexes
+  ) AS json_parsed
 JOIN
   -- Unnest the JSON-parsed tlds.
-  UNNEST(regexes.tlds) AS tld
+  UNNEST(json_parsed.tlds) AS tld
 -- Exclude cases that can't be tabulated correctly, where activityReportField
 -- is null/empty, or TLD is null/empty despite being a domain flow.
 WHERE