fix: scrub invalid UTF-8 in CloudFront log lines

CHANGELOG.md

@@ -1,3 +1,6 @@
+## Release 0.2.3
+- [fix] parse CloudFront lines as binary and normalize fields to UTF-8 (ISO-8859-1 fallback) so Latin-1 percent-encoding like `%E9` and invalid UTF-8 no longer break `CGI.unescape`/`.split` or Loki JSON
+
 ## Release 0.2.2
 - [end] replace slow gsub with more efficient string replace
 

fluent-plugin-cloudfront-log-optimized.gemspec

@@ -4,7 +4,7 @@ $LOAD_PATH.unshift(lib) unless $LOAD_PATH.include?(lib)
 
 Gem::Specification.new do |spec|
   spec.name          = "fluent-plugin-cloudfront-log-optimized"
-  spec.version       = "0.2.2"
+  spec.version       = "0.2.3"
   spec.authors       = ["kubihee", "lenfree", "kjwierenga"]
   spec.email         = ["kubihie@gmail.com", "lenfree.yeung@gmail.com", "k.j.wierenga@gmail.com"]
 

lib/fluent/plugin/in_cloudfront_log.rb

@@ -29,6 +29,7 @@ def initialize
     require 'aws-sdk-s3'
     require 'time'
     require 'uri'
+    require 'cgi'
   end
 
   def configure(conf)
@@ -126,9 +127,24 @@ def purge(filename)
     end
   end
 
+  # Normalize field values to valid UTF-8 for Loki JSON.
+  # Prefer keeping already-valid UTF-8; otherwise treat bytes as ISO-8859-1
+  # (common for percent-encoded Latin-1 in query strings, e.g. %E9 → é).
+  def scrub_utf8(value)
+    s = value.to_s.dup
+    s.force_encoding('UTF-8')
+    return s if s.valid_encoding?
+
+    s.force_encoding('ISO-8859-1').encode('UTF-8')
+  end
+
   def process_line(line)
+    # Parse as binary: CGI.unescape of %E9/%FF etc. yields high bytes that are
+    # invalid when the string is tagged UTF-8, and String#split then raises.
+    line = line.to_s.dup.force_encoding('ASCII-8BIT')
+
     if line[0.1] == '#'
-      parse_header(line)
+      parse_header(scrub_utf8(line))
       return
     end
 
@@ -140,6 +156,10 @@ def process_line(line)
       CGI.unescape(line).split("\t")
     ].transpose.to_h
 
+    record.each do |key, value|
+      record[key] = scrub_utf8(value) if value.is_a?(String)
+    end
+
     timestamp = if @parse_date_time
       Time.iso8601("#{record['date']}T#{record['time']}+00:00").to_i
     else

test/plugin/test_in_cloudfrontlog.rb

@@ -139,4 +139,63 @@ def create_driver(conf = MINIMAL_CONFIG)
     end
   end
 
+  sub_test_case "invalid UTF-8 in log lines" do
+    FIELDS_LINE = "#Fields: date time x-edge-location sc-bytes c-ip cs-method cs(Host) cs-uri-stem sc-status cs(Referer) cs(User-Agent) cs-uri-query cs(Cookie) x-edge-result-type x-edge-request-id x-host-header cs-protocol cs-bytes time-taken x-forwarded-for ssl-protocol ssl-cipher x-edge-response-result-type cs-protocol-version fle-status fle-encrypted-fields c-port time-to-first-byte x-edge-detailed-result-type sc-content-type sc-content-len sc-range-start sc-range-end"
+
+    def build_line(uri_query:, user_agent: 'Mozilla/5.0')
+      [
+        '2026-08-07', '19:49:19', 'IAD55-C1', '1234', '1.2.3.4', 'GET', 'd111.cloudfront.net',
+        '/path', '200', 'https://example.com/', user_agent, uri_query, '-', 'Hit', 'AbCdEf',
+        'www.example.com', 'https', '200', '0.050', '-', 'TLSv1.2', 'ECDHE-RSA-AES128-GCM-SHA256',
+        'Hit', 'HTTP/2.0', '-', '-', '54321', '0.010', 'Hit', 'text/html', '1234', '-', '-'
+      ].join("\t")
+    end
+
+    def prime_and_process(instance, line)
+      instance.process_line("#Version: 1.0")
+      instance.process_line(FIELDS_LINE)
+      instance.process_line(line)
+    end
+
+    test "scrubs invalid UTF-8 bytes without raising" do
+      driver = create_driver(MINIMAL_CONFIG)
+      line = build_line(uri_query: "q=bad\xFFbyte&x=1", user_agent: "Mozilla/\x80Bot")
+
+      emitted_event = nil
+      assert_nothing_raised {
+        emitted_event = prime_and_process(driver.instance, line)
+      }
+
+      assert_equal(true, emitted_event['cs-uri-query'].encoding == Encoding::UTF_8)
+      assert_equal(true, emitted_event['cs-uri-query'].valid_encoding?)
+      assert_equal(true, emitted_event['cs(User-Agent)'].valid_encoding?)
+      assert_equal(true, emitted_event['cs-uri-query'].include?('q=bad'))
+      assert_equal(true, emitted_event['cs(User-Agent)'].include?('Mozilla/'))
+    end
+
+    test "preserves valid Unicode percent-encoding" do
+      driver = create_driver(MINIMAL_CONFIG)
+      line = build_line(uri_query: 'q=%C3%A7af%C3%A9&name=Jo%C3%A3o')
+
+      emitted_event = prime_and_process(driver.instance, line)
+
+      assert_equal(true, emitted_event['cs-uri-query'].include?('çafé'))
+      assert_equal(true, emitted_event['cs-uri-query'].include?('João'))
+    end
+
+    test "decodes Latin-1 percent-encoding without raising" do
+      driver = create_driver(MINIMAL_CONFIG)
+      # Real CloudFront case: "métricas" encoded as Latin-1 %E9 instead of UTF-8 %C3%A9
+      line = build_line(uri_query: 'q=5%20m%E9tricas%20que%20te%20pueden%20servir')
+
+      emitted_event = nil
+      assert_nothing_raised {
+        emitted_event = prime_and_process(driver.instance, line)
+      }
+
+      assert_equal(true, emitted_event['cs-uri-query'].valid_encoding?)
+      assert_equal(true, emitted_event['cs-uri-query'].include?('métricas'))
+    end
+  end
+
 end