CVE-2026-8838

@@ -83,7 +83,10 @@ def int2_recv(data: bytes, offset: int, length: int) -> int:
 
 
 def vector_in(data: bytes, idx: int, length: int) -> typing.List:
-    return eval("[" + data[idx : idx + length].decode(_client_encoding).replace(" ", ",") + "]")
+    text = data[idx : idx + length].decode(_client_encoding).strip()
+    if not text:
+        return []
+    return [int(x) for x in text.split()]
 
 
 def int4_recv(data: bytes, offset: int, length: int) -> int:
@@ -222,26 +225,6 @@ def intervald2s_send_integer(v: IntervalDayToSecond) -> bytes:
     return typing.cast(bytes, q_pack(microseconds))
 
 
-glbls: typing.Dict[str, type] = {"Decimal": Decimal}
-trans_tab = dict(zip(map(ord, "{}"), "[]"))
-
-
-# def array_in(data: bytes, idx: int, length: int) -> typing.List:
-#     arr: typing.List[str] = []
-#     prev_c = None
-#     for c in data[idx:idx + length].decode(
-#             _client_encoding).translate(
-#         trans_tab).replace('NULL', 'None'):
-#         if c not in ('[', ']', ',', 'N') and prev_c in ('[', ','):
-#             arr.extend("Decimal('")
-#         elif c in (']', ',') and prev_c not in ('[', ']', ',', 'e'):
-#             arr.extend("')")
-#
-#         arr.append(c)
-#         prev_c = c
-#     return typing.cast(typing.List, eval(''.join(arr), glbls))
-
-
 def numeric_in_binary(data: bytes, offset: int, length: int, scale: int) -> Decimal:
     raw_value: int
 

@@ -84,3 +84,44 @@ def test_timestamp_recv_integer(_input) -> None:
     print(type_utils.timestamp_recv_integer(in_val, 0, 0))
     print(EPOCH.timestamp() * 1000)
     assert type_utils.timestamp_recv_integer(in_val, 0, 0) == exp_val
+
+
+# --- vector_in tests ---
+
+vector_in_normal_data: typing.List[typing.Tuple[bytes, typing.List[int]]] = [
+    (b"1 2 3", [1, 2, 3]),
+    (b"1", [1]),
+    (b"0", [0]),
+    (b"-1", [-1]),
+    (b"-32768", [-32768]),
+    (b"32767", [32767]),
+    (b"-1 2 -3", [-1, 2, -3]),
+    (b"1 0 3", [1, 0, 3]),
+    (b"", []),
+]
+
+
+@pytest.mark.parametrize("_input", vector_in_normal_data)
+def test_vector_in_normal(_input) -> None:
+    """Verify vector_in correctly parses space-separated int16 values."""
+    data, expected = _input
+    result = type_utils.vector_in(data, 0, len(data))
+    assert result == expected
+    for val in result:
+        assert isinstance(val, int)
+
+
+vector_in_injection_payloads: typing.List[bytes] = [
+    b"__import__('os').system('echo exploited')",
+    b"__import__('os').uname().sysname",
+    b"__import__('subprocess').check_output(['whoami'])",
+    b"open('/etc/passwd').read()",
+    b"exec('import os')",
+]
+
+
+@pytest.mark.parametrize("payload", vector_in_injection_payloads)
+def test_vector_in_rejects_code_injection(payload) -> None:
+    """Verify vector_in raises an exception for non-integer input instead of executing it."""
+    with pytest.raises((ValueError, TypeError)):
+        type_utils.vector_in(payload, 0, len(payload))