diff --git a/cloudinit/config/cc_write_files.py b/cloudinit/config/cc_write_files.py
index e07bc2c7189..7cc7f854bb4 100644
--- a/cloudinit/config/cc_write_files.py
+++ b/cloudinit/config/cc_write_files.py
@@ -18,7 +18,7 @@
 DEFAULT_OWNER = "root:root"
 DEFAULT_PERMS = 0o644
 DEFAULT_DEFER = False
-UNKNOWN_ENC = "text/plain"
+TEXT_PLAIN_ENC = "text/plain"
 
 LOG = logging.getLogger(__name__)
 
@@ -141,14 +141,18 @@ def canonicalize_extraction(encoding_type):
     # Yaml already encodes binary data as base64 if it is given to the
     # yaml file as binary, so those will be automatically decoded for you.
     # But the above b64 is just for people that are more 'comfortable'
-    # specifing it manually (which might be a possiblity)
+    # specifing it manually (which might be a possibility)
     if encoding_type in ["b64", "base64"]:
         return ["application/base64"]
+    if encoding_type == TEXT_PLAIN_ENC:
+        return [TEXT_PLAIN_ENC]
     if encoding_type:
         LOG.warning(
-            "Unknown encoding type %s, assuming %s", encoding_type, UNKNOWN_ENC
+            "Unknown encoding type %s, assuming %s",
+            encoding_type,
+            TEXT_PLAIN_ENC,
         )
-    return [UNKNOWN_ENC]
+    return [TEXT_PLAIN_ENC]
 
 
 def write_files(name, files):
@@ -202,7 +206,7 @@ def extract_contents(contents, extraction_types):
             result = util.decomp_gzip(result, quiet=False, decode=False)
         elif t == "application/base64":
             result = base64.b64decode(result)
-        elif t == UNKNOWN_ENC:
+        elif t == TEXT_PLAIN_ENC:
             pass
     return result
 
diff --git a/cloudinit/config/schemas/schema-cloud-config-v1.json b/cloudinit/config/schemas/schema-cloud-config-v1.json
index 67e1a5bb5b3..d409d5d678b 100644
--- a/cloudinit/config/schemas/schema-cloud-config-v1.json
+++ b/cloudinit/config/schemas/schema-cloud-config-v1.json
@@ -2105,7 +2105,7 @@
               "encoding": {
                 "type": "string",
                 "default": "text/plain",
-                "enum": ["gz", "gzip", "gz+base64", "gzip+base64", "gz+b64", "gzip+b64", "b64", "base64"],
+                "enum": ["gz", "gzip", "gz+base64", "gzip+base64", "gz+b64", "gzip+b64", "b64", "base64", "text/plain"],
                 "description": "Optional encoding type of the content. Default is ``text/plain`` and no content decoding is performed. Supported encoding types are: gz, gzip, gz+base64, gzip+base64, gz+b64, gzip+b64, b64, base64"
               },
               "append": {
diff --git a/tests/unittests/config/test_cc_write_files.py b/tests/unittests/config/test_cc_write_files.py
index 956e4327579..01c920e8bae 100644
--- a/tests/unittests/config/test_cc_write_files.py
+++ b/tests/unittests/config/test_cc_write_files.py
@@ -139,6 +139,27 @@ def test_all_decodings(self):
         )
         self.assertEqual(len(expected), flen_expected)
 
+    def test_handle_plain_text(self):
+        self.patchUtils(self.tmp)
+        file_path = "/tmp/file-text-plain"
+        content = "asdf"
+        cfg = {
+            "write_files": [
+                {
+                    "content": content,
+                    "path": file_path,
+                    "encoding": "text/plain",
+                    "defer": False,
+                }
+            ]
+        }
+        cc = self.tmp_cloud("ubuntu")
+        handle("ignored", cfg, cc, LOG, [])
+        assert content == util.load_file(file_path)
+        self.assertNotIn(
+            "Unknown encoding type text/plain", self.logs.getvalue()
+        )
+
     def test_deferred(self):
         self.patchUtils(self.tmp)
         file_path = "/tmp/deferred.file"
@@ -213,11 +234,29 @@ class TestWriteFilesSchema:
                     "write_files.0.encoding: 'g' is not one of ['gz', 'gzip',"
                 ),
             ),
+            (
+                {
+                    "write_files": [
+                        {
+                            "append": False,
+                            "content": "a",
+                            "encoding": "text/plain",
+                            "owner": "jeff",
+                            "path": "/some",
+                            "permissions": "0777",
+                        }
+                    ]
+                },
+                None,
+            ),
         ],
     )
     @skipUnlessJsonSchema()
     def test_schema_validation(self, config, error_msg):
-        with pytest.raises(SchemaValidationError, match=error_msg):
+        if error_msg is not None:
+            with pytest.raises(SchemaValidationError, match=error_msg):
+                validate_cloudconfig_schema(config, get_schema(), strict=True)
+        else:
             validate_cloudconfig_schema(config, get_schema(), strict=True)