Protect against overflow/underflow for integer record types

CHANGELOG.rst

@@ -9,7 +9,7 @@ Versioning <https://semver.org/spec/v2.0.0.html>`_.
 
 Unreleased_
 -----------
-
+- `Protect against overflow/underflow for integer record types <../../pull/110>`_
 - `Allow "status" and "severity" on In record init <../../pull/111>`_
 
 4.1.0_ - 2022-08-05

softioc/device.py

@@ -188,6 +188,20 @@ def __completion(self, record):
         if self._blocking:
             signal_processing_complete(record, self._callback)
 
+    def _validate_value(self, new_value):
+        """Checks whether the new value is valid; if so, returns True"""
+        try:
+            self._value_to_epics(new_value)
+        except AssertionError:
+            return False
+        if (
+            self.__enable_write
+            and self.__validate
+            and not self.__validate(self, new_value)
+        ):
+            return False
+        return True
+
     def _process(self, record):
         '''Processing suitable for output records.  Performs immediate value
         validation and asynchronous update notification.'''
@@ -202,8 +216,7 @@ def _process(self, record):
             return EPICS_OK
 
         python_value = self._epics_to_value(value)
-        if self.__enable_write and self.__validate and \
-                not self.__validate(self, python_value):
+        if not self._validate_value(python_value):
             # Asynchronous validation rejects value, so restore the last good
             # value.
             self._write_value(record, self._value)
@@ -252,8 +265,17 @@ def get(self):
         return self._epics_to_value(value)
 
 
-def _Device(Base, record_type, ctype, dbf_type, epics_rc, mlst = False):
-    '''Wrapper for generating simple records.'''
+def _Device(
+    Base,
+    record_type,
+    ctype,
+    dbf_type,
+    epics_rc,
+    value_valid,
+    mlst=False,
+):
+    """Wrapper for generating simple records."""
+
     class GenericDevice(Base):
         _record_type_ = record_type
         _device_name_ = 'devPython_' + record_type
@@ -264,6 +286,13 @@ class GenericDevice(Base):
         if mlst:
             _fields_.append('MLST')
 
+        def _value_to_epics(self, value):
+            assert value_valid(value), (
+                f"Value {value} out of valid range for record type "
+                f"{self._record_type_}"
+            )
+            return super()._value_to_epics(value)
+
     GenericDevice.__name__ = record_type
     return GenericDevice
 
@@ -276,13 +305,36 @@ def _Device_In(*args, **kargs):
 def _Device_Out(*args, **kargs):
     return _Device(_Out, mlst = True, *args, **kargs)
 
-longin = _Device_In('longin', c_int32, fields.DBF_LONG, EPICS_OK)
-longout = _Device_Out('longout', c_int32, fields.DBF_LONG, EPICS_OK)
-bi = _Device_In('bi', c_uint16, fields.DBF_CHAR, NO_CONVERT)
-bo = _Device_Out('bo', c_uint16, fields.DBF_CHAR, NO_CONVERT)
-mbbi = _Device_In('mbbi', c_uint16, fields.DBF_SHORT, NO_CONVERT)
-mbbo = _Device_Out('mbbo', c_uint16, fields.DBF_SHORT, NO_CONVERT)
+_long_min = numpy.iinfo(numpy.int32).min
+_long_max = numpy.iinfo(numpy.int32).max
 
+longin = _Device_In(
+    "longin",
+    c_int32,
+    fields.DBF_LONG,
+    EPICS_OK,
+    lambda x: _long_min <= x <= _long_max,
+)
+longout = _Device_Out(
+    "longout",
+    c_int32,
+    fields.DBF_LONG,
+    EPICS_OK,
+    lambda x: _long_min <= x <= _long_max,
+)
+
+bi = _Device_In(
+    "bi", c_uint16, fields.DBF_CHAR, NO_CONVERT, lambda x: 0 <= x <= 1
+)
+bo = _Device_Out(
+    "bo", c_uint16, fields.DBF_CHAR, NO_CONVERT, lambda x: 0 <= x <= 1
+)
+mbbi = _Device_In(
+    "mbbi", c_uint16, fields.DBF_SHORT, NO_CONVERT, lambda x: 0 <= x <= 15
+)
+mbbo = _Device_Out(
+    "mbbo", c_uint16, fields.DBF_SHORT, NO_CONVERT, lambda x: 0 <= x <= 15
+)
 
 def _string_at(value, count):
     # Need string_at() twice to ensure string is size limited *and* null

softioc/extension.c

@@ -69,8 +69,13 @@ static PyObject *get_field_offsets(PyObject *self, PyObject *args)
 
     int status = dbFindRecordType(&dbentry, record_type);
     if (status != 0)
-        printf("Unable to find record type \"%s\" (error %d)\n",
+    {
+        return PyErr_Format(
+            PyExc_RuntimeError,
+            "Unable to find record type \"%s\" (error %d)\n",
             record_type, status);
+        Py_RETURN_NONE;
+    }
     else
         status = dbFirstField(&dbentry, 0);
 

tests/test_record_values.py

@@ -12,10 +12,12 @@
     log,
     select_and_recv,
     TIMEOUT,
-    get_multiprocessing_context
+    get_multiprocessing_context,
+    create_random_prefix
 )
 
 from softioc import asyncio_dispatcher, builder, softioc
+from softioc.fields import DBR_STRING
 from softioc.pythonSoftIoc import RecordWrapper
 
 # Test file for anything related to valid record values getting/setting
@@ -885,3 +887,217 @@ def test_waveform_rejects_overlong_values(self):
             w_in.set([1, 2, 3, 4])
         with pytest.raises(AssertionError):
             w_out.set([1, 2, 3, 4])
+
+    def test_long_rejects_invalid_values(self):
+        """Test that longIn/longOut reject invalid values"""
+        l_in = builder.longIn("L_IN_1")
+        l_out = builder.longOut("L_OUT_1")
+
+        long_min = numpy.iinfo(numpy.int32).min
+        long_max = numpy.iinfo(numpy.int32).max
+
+        with pytest.raises(AssertionError):
+            l_in.set(long_min - 1)
+        with pytest.raises(AssertionError):
+            l_out.set(long_min - 1)
+
+        with pytest.raises(AssertionError):
+            l_in.set(long_max + 1)
+        with pytest.raises(AssertionError):
+            l_out.set(long_max + 1)
+
+        # And confirm that creating with invalid values also raises
+        with pytest.raises(AssertionError):
+            builder.longIn("L_IN_2", initial_value = long_min - 1)
+        with pytest.raises(AssertionError):
+            builder.longOut("L_OUT_2", initial_value = long_min - 1)
+
+        with pytest.raises(AssertionError):
+            builder.longIn("L_IN_3", initial_value = long_max + 1)
+        with pytest.raises(AssertionError):
+            builder.longOut("L_OUT_3", initial_value = long_max + 1)
+
+    def test_bool_rejects_invalid_values(self):
+        """Test that boolIn/boolOut rejects invalid values"""
+        b_in = builder.boolIn("B_IN_1")
+        b_out = builder.boolOut("B_OUT_1")
+
+        with pytest.raises(AssertionError):
+            b_in.set(-1)
+        with pytest.raises(AssertionError):
+            b_out.set(-1)
+
+        with pytest.raises(AssertionError):
+            b_in.set(2)
+        with pytest.raises(AssertionError):
+            b_out.set(2)
+
+        # And confirm that creating with invalid values also raises
+        with pytest.raises(AssertionError):
+            builder.boolIn("B_IN_2", initial_value=-1)
+        with pytest.raises(AssertionError):
+            builder.boolOut("B_OUT_2", initial_value=-1)
+
+        with pytest.raises(AssertionError):
+            builder.boolIn("B_IN_3", initial_value=2)
+        with pytest.raises(AssertionError):
+            builder.boolOut("B_OUT_3", initial_value=2)
+
+    def test_mbb_rejects_invalid_values(self):
+        """Test that mbbIn/mbbOut rejects invalid values"""
+        mbb_in = builder.mbbIn("MBB_IN_1")
+        mbb_out = builder.mbbOut("MBB_OUT_1")
+
+        with pytest.raises(AssertionError):
+            mbb_in.set(-1)
+        with pytest.raises(AssertionError):
+            mbb_out.set(-1)
+
+        with pytest.raises(AssertionError):
+            mbb_in.set(16)
+        with pytest.raises(AssertionError):
+            mbb_out.set(16)
+
+        # And confirm that creating with invalid values also raises
+        with pytest.raises(AssertionError):
+            builder.mbbIn("MBB_IN_2", initial_value=-1)
+        with pytest.raises(AssertionError):
+            builder.mbbOut("MBB_OUT_2", initial_value=-1)
+
+        with pytest.raises(AssertionError):
+            builder.mbbIn("MBB_IN_3", initial_value=16)
+        with pytest.raises(AssertionError):
+            builder.mbbOut("MBB_OUT_3", initial_value=16)
+
+    def invalid_value_test_func(self, device_name, conn, creation_func):
+
+        builder.SetDeviceName(device_name)
+
+        creation_func("INVALID_VAL_REC")
+
+        dispatcher = asyncio_dispatcher.AsyncioDispatcher()
+        builder.LoadDatabase()
+        softioc.iocInit(dispatcher)
+
+        conn.send("R")  # "Ready"
+
+        log("CHILD: Sent R over Connection to Parent")
+
+        # Keep process alive while main thread runs CAGET
+        if conn.poll(TIMEOUT):
+            val = conn.recv()
+            assert val == "D", "Did not receive expected Done character"
+
+        log("CHILD: Received exit command, child exiting")
+
+    @requires_cothread
+    @pytest.mark.parametrize(
+        "creation_func, invalid_min, invalid_max, expected_val",
+        [
+            (
+                builder.longOut,
+                numpy.iinfo(numpy.int32).min - 1,
+                numpy.iinfo(numpy.int32).max + 1,
+                0,
+            ),
+            (
+                builder.boolOut,
+                -1,
+                2,
+                0,
+            ),
+            (
+                builder.mbbOut,
+                -1,
+                16,
+                0,
+            ),
+        ],
+
+    )
+    def test_invalid_values_caput(
+        self, creation_func, invalid_min, invalid_max, expected_val
+    ):
+        """Test that attempting to set invalid values causes caput to return
+        an error, and the record's value remains unchanged."""
+        if creation_func == builder.mbbOut:
+            pytest.skip(
+                "Bug somewhere, possibly Cothread, means that errors are not "
+                "reported for mbbOut records")
+
+        ctx = get_multiprocessing_context()
+
+        parent_conn, child_conn = ctx.Pipe()
+
+        device_name = create_random_prefix()
+
+        process = ctx.Process(
+            target=self.invalid_value_test_func,
+            args=(device_name, child_conn, creation_func),
+        )
+
+        process.start()
+
+        log("PARENT: Child started, waiting for R command")
+
+        from cothread.catools import caget, caput, _channel_cache
+
+        try:
+            # Wait for message that IOC has started
+            select_and_recv(parent_conn, "R")
+
+            log("PARENT: received R command")
+
+            # Suppress potential spurious warnings
+            _channel_cache.purge()
+
+            record_name = device_name + ":INVALID_VAL_REC"
+
+            log(f"PARENT: Putting invalid min value {invalid_min} to record")
+            # Send as string data, otherwise catools will automatically convert
+            # it to a valid int32
+            put_ret = caput(
+                record_name,
+                invalid_min,
+                wait=True,
+                datatype=DBR_STRING,
+                throw=False,
+            )
+            assert not put_ret.ok, \
+                "caput for invalid minimum value unexpectedly succeeded"
+
+            log(f"PARENT: Putting invalid max value {invalid_max} to record")
+            put_ret = caput(
+                record_name,
+                invalid_max,
+                wait=True,
+                datatype=DBR_STRING,
+                throw=False,
+            )
+            assert not put_ret.ok, \
+                "caput for invalid maximum value unexpectedly succeeded"
+
+
+            log("PARENT: Getting value from record")
+
+            ret_val = caget(
+                record_name,
+                timeout=TIMEOUT,
+            )
+            assert ret_val.ok, \
+                f"caget did not succeed: {ret_val.errorcode}, {ret_val}"
+
+            log(f"PARENT: Received val from record: {ret_val}")
+
+            assert ret_val == expected_val
+
+        finally:
+            # Suppress potential spurious warnings
+            _channel_cache.purge()
+
+            log("PARENT: Sending Done command to child")
+            parent_conn.send("D")  # "Done"
+            process.join(timeout=TIMEOUT)
+            log(f"PARENT: Join completed with exitcode {process.exitcode}")
+            if process.exitcode is None:
+                pytest.fail("Process did not terminate")

tests/test_records.py

@@ -473,9 +473,14 @@ def on_update_runner(self, creation_func, always_update, put_same_value):
             log("PARENT: begining While loop")
 
             while count < 4:
+
+                new_val = count
+                if creation_func == builder.boolOut:
+                    new_val = count % 2
+
                 put_ret = caput(
                     device_name + ":ON-UPDATE-RECORD",
-                    9 if put_same_value else count,
+                    1 if put_same_value else new_val,
                     wait=True,
                 )
                 assert put_ret.ok, f"caput did not succeed: {put_ret.errorcode}"