problem_report: enforce some keys to always have str values in load()

problem_report.py

@@ -9,6 +9,8 @@
 # option) any later version.  See http://www.gnu.org/copyleft/gpl.html for
 # the full text of the license.
 
+# pylint: disable=too-many-lines
+
 import base64
 import binascii
 import collections
@@ -352,6 +354,64 @@ def splitlines(self) -> list[bytes]:
 
 
 ProblemReportValue: TypeAlias = bytes | CompressedFile | CompressedValue | str | tuple
+_STRING_KEYS = {
+    "Annotation",
+    "Architecture",
+    "AssertionMessage",
+    "AuditLog",
+    "CheckboxSubmission",
+    "CrashCounter",
+    "CrashDB",
+    "Date",
+    "DbusErrorAnalysis",
+    "Dependencies",
+    "DesktopFile",
+    "DialogBody",
+    "Disassembly",
+    "DistroRelease",
+    "DuplicateSignature",
+    "ExecutablePath",
+    "ExecutableTimestamp",
+    "Failure",
+    "GLibAssertionMessage",
+    "InterpreterPath",
+    "KernLog",
+    "MachineType",
+    "MainClassUrl",
+    "NonfreeKernelModules",
+    "OopsText",
+    "OpenFds",
+    "Package",
+    "PackageArchitecture",
+    "ProblemType",
+    "ProcCmdline",
+    "ProcCwd",
+    "ProcEnviron",
+    "ProcMaps",
+    "ProcStatus",
+    "Registers",
+    "RespawnCommand",
+    "SegvAnalysis",
+    "SegvAnalysisError",
+    "Signal",
+    "SignalName",
+    "SnapTags",
+    "SourcePackage",
+    "StackTrace",
+    "Stacktrace",
+    "StacktraceSource",
+    "StacktraceTop",
+    "Tags",
+    "ThreadStacktrace",
+    "Title",
+    "Traceback",
+    "Uname",
+    "UnreportableReason",
+    "UserGroups",
+    "_KnownReport",
+    "_PythonExceptionQualifier",
+    "dmi.bios.version",
+}
 
 
 class ProblemReport(collections.UserDict):
@@ -419,12 +479,11 @@ def load(
                         name=key, compressed_value=b"".join(iterator)
                     )
                 else:
-                    self.data[key] = self._try_unicode(
-                        b"".join(CompressedValue.decode_compressed_stream(iterator))
-                    )
+                    value = b"".join(CompressedValue.decode_compressed_stream(iterator))
+                    self.data[key] = self._try_unicode(key, value)
 
             else:
-                self.data[key] = self._try_unicode(b"".join(iterator))
+                self.data[key] = self._try_unicode(key, b"".join(iterator))
 
             if remaining_keys is not None:
                 remaining_keys.remove(key)
@@ -516,8 +575,10 @@ def is_binary(string: bytes | str) -> bool:
         return False
 
     @classmethod
-    def _try_unicode(cls, value: bytes) -> bytes | str:
+    def _try_unicode(cls, key: str, value: bytes) -> bytes | str:
         """Try to convert bytearray value to Unicode."""
+        if key in _STRING_KEYS:
+            return value.decode("UTF-8")
         if not cls.is_binary(value):
             try:
                 return value.decode("UTF-8")

tests/unit/test_problem_report.py

@@ -229,6 +229,20 @@ def test_load(self) -> None:
         pr.load(io.BytesIO(b"ProblemType: Crash"))
         self.assertEqual(list(pr.keys()), ["ProblemType"])
 
+    def test_load_string_with_null(self) -> None:
+        """Test load() with a NULL character in a string key.
+
+        Test case for https://launchpad.net/bugs/2146806
+        """
+        proc_maps = "7f8e5d61f000-7f8e5d620000 r--p 00005000 fc:00 190815U\0\0"
+        content = f"ProblemType: Crash\n" f"Date: now!\n" f"ProcMaps: {proc_maps}\n"
+        report = problem_report.ProblemReport()
+        report.load(io.BytesIO(content.encode()))
+
+        self.assertEqual(report["ProblemType"], "Crash")
+        self.assertEqual(report["Date"], "now!")
+        self.assertEqual(report["ProcMaps"], proc_maps)
+
     def test_load_binary_blob(self) -> None:
         """Throw exception when binary file (e.g. core) is loaded."""
         report = problem_report.ProblemReport()