jstaylor11 · Oct 15, 2026 · Oct 15, 2026 · Oct 16, 2026
diff --git a/edlclient/Library/Modules/oneplus_param.py b/edlclient/Library/Modules/oneplus_param.py
@@ -378,8 +378,7 @@ def parse_encrypted(self, rdata, sid):
         itemdata, hv, cv, updatecounter = self.decryptsid(data)
         if itemdata is not None:
             itemdata = bytearray(itemdata)
-            print(
-                f"Offset {hex(sid * 0x400)}: hv {hex(hv)}, cv {hex(cv)}, increase_enc_update_counter {hex(updatecounter)}.")
+            print(f"Offset {hex(sid * 0x400)}: hv {hex(hv)}, cv {hex(cv)}, increase_enc_update_counter {hex(updatecounter)}.")
             i = 0
             while i < len(itemdata):
                 offset = i + 0x80
@@ -484,29 +483,35 @@ def parse_data(self, i, itemdata, offset, param, sidindex, encrypted=False):
         return length
 
     def setparamvalue(self, data, sid, offset, value):
+        data = bytearray(data)
+
         if sid > 0x100:
             rdata = data[sid * 0x400:(sid * 0x400) + 0x1000]
             itemdata, hv, cv, updatecounter = self.decryptsid(rdata)
-            if itemdata is not None:
-                itemdata = bytearray(itemdata)
-                if isinstance(value, int):
-                    itemdata[offset - 0x80:(offset + 4) - 0x80] = pack("<I", value)
-                elif isinstance(value, bytearray):
-                    itemdata[offset - 0x80:(offset + len(value)) - 0x80] = value
-                # itemdata[0x84-0x80:(0x84+4)-0x80]=pack("<I",0xB7) #devkmsg_enable
-                mdata = self.encryptsid(itemdata, hv, cv, updatecounter)
-            data = bytearray(data)
-            data[sid * 0x400:(sid * 0x400) + 0x1000] = mdata
-            # data[(sid+0x200) * 0x400:((sid+0x200) * 0x400) + 0x1000] = mdata
-        else:
-            rdata = data[sid * 0x400:(sid * 0x400) + 0x1000]
-            data = bytearray(data)
-            rdata = bytearray(rdata)
+            if itemdata is None:
+                raise ValueError(f"Cannot modify SID {sid:#x}: decryption or integrity validation failed")
+
+            itemdata = bytearray(itemdata)
             if isinstance(value, int):
-                rdata[offset:offset + 4] = pack("<I", value)
+                itemdata[offset - 0x80:(offset + 4) - 0x80] = pack("<I", 
8000
value)
             elif isinstance(value, bytearray):
-                rdata[offset:(offset + len(value))] = value
-            data[sid * 0x400:(sid * 0x400) + 0x1000] = rdata
+                itemdata[offset - 0x80:(offset + len(value)) - 0x80] = value
+            else:
+                raise TypeError("Unsupported value type")
+
+            mdata = self.encryptsid(itemdata, hv, cv, updatecounter)
+            data[sid * 0x400:(sid * 0x400) + 0x1000] = mdata
+            return data
+
+        rdata = bytearray(data[sid * 0x400:(sid * 0x400) + 0x1000])
+        if isinstance(value, int):
+            rdata[offset:offset + 4] = pack("<I", value)
+        elif isinstance(value, bytearray):
+            rdata[offset:(offset + len(value))] = value
+        else:
+            raise TypeError("Unsupported value type")
+
+        data[sid * 0x400:(sid * 0x400) + 0x1000] = rdata
         return data
 
     def gencode(self, inparray):
@@ -969,8 +974,10 @@ def main():
         param = paramtools(mode, serial)
         with open(filename, 'rb') as rf:
             data = rf.read()
+            # Validate before creating or truncating the output file. (Openning file in wb mode trunicates it immediately, before wf.write() evaluates its argument.
+            patched = param.setparamvalue(data, sid, offset, value)
             with open(filename + ".patched", 'wb') as wf:
-                wf.write(param.setparamvalue(data, sid, offset, value))
+                wf.write(patched)
     elif args["gencode"]:
         imei = args["<imei>"]
         mode = 0

diff --git a/edlclient/Library/TestFiles/test_oneplus_param.py b/edlclient/Library/TestFiles/test_oneplus_param.py
@@ -0,0 +1,71 @@
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from edlclient.Library.Modules import oneplus_param as op
+
+
+class ParamWriteTests(unittest.TestCase):
+    def test_failed_decryption_does_not_mutate_input(self):
+        tool = op.paramtools(0, 0)
+        data = bytearray(0x50000)
+        before = bytes(data)
+        with patch.object(tool, "decryptsid", return_value=(None,) * 4):
+            with patch.object(tool, "encryptsid") as encrypt:
+                with self.assertRaisesRegex(ValueError, "SID 0x13c"):
+                    tool.setparamvalue(data, 0x13C, 0x80, 1)
+                encrypt.assert_not_called()
+        self.assertEqual(bytes(data), before)
+
+    def test_valid_encrypted_record_still_updates(self):
+        tool = op.paramtools(0, 0)
+        base = 0x12C * 0x400
+        data = bytearray(base + 0x1000)
+        data[base:] = tool.encryptsid(bytes(0xB80), 1, 1, 0)
+        before = bytes(data)
+        result = tool.setparamvalue(data, 0x12C, 0x80, 7)
+        item, hv, cv, counter = tool.decryptsid(result[base:])
+        self.assertIsNotNone(item)
+        assert item is not None
+        self.assertEqual(item[:4], b"\x07\x00\x00\x00")
+        self.assertEqual(item[4:], bytes(0xB80 - 4))
+        self.assertEqual((hv, cv, counter), (1, 1, 2))
+        self.assertEqual(result[:base], before[:base])
+        self.assertEqual(len(result), len(before))
+        self.assertEqual(bytes(data), before)
+
+    def test_cli_failure_preserves_output_state(self):
+        for existing in (False, True):
+            with self.subTest(existing=existing):
+                with tempfile.TemporaryDirectory() as folder:
+                    source = Path(folder) / "param.bin"
+                    source.write_bytes(bytes(0x50000))
+                    output = Path(str(source) + ".patched")
+                    if existing:
+                        output.write_bytes(b"keep this file")
+                    args = [
+                        "oneplus_param.py",
+                        "setparam",
+                        str(source),
+                        "0x13C",
+                        "0x80",
+                        "1",
+                    ]
+                    with patch("sys.argv", args):
+                        with patch.object(
+                            op.paramtools,
+                            "decryptsid",
+                            return_value=(None,) * 4,
+                        ):
+                            with self.assertRaises(ValueError):
+                                op.main()
+                    self.assertEqual(source.read_bytes(), bytes(0x50000))
+                    if existing:
+                        self.assertEqual(output.read_bytes(), b"keep this file")
+                    else:
+                        self.assertFalse(output.exists())
+
+
+if __name__ == "__main__":
+    unittest.main()