diff --git a/src/middlewared/middlewared/plugins/iscsi_/extents.py b/src/middlewared/middlewared/plugins/iscsi_/extents.py
old mode 100755
new mode 100644
index 6087b0d51c39..080e3f3195c4
--- a/src/middlewared/middlewared/plugins/iscsi_/extents.py
+++ b/src/middlewared/middlewared/plugins/iscsi_/extents.py
@@ -104,8 +104,8 @@ async def do_create(self, data):
 
         `ro` when set to true prevents the initiator from writing to this LUN.
         """
+        await self.validate(data)
         verrors = ValidationErrors()
-        await self.middleware.call('iscsi.extent.validate', data)
         await self.clean(data, 'iscsi_extent_create', verrors)
         verrors.check()
 
@@ -114,7 +114,16 @@ async def do_create(self, data):
         # This change is being made in conjunction with threads_num being specified in scst.conf
         if data['type'] == 'DISK' and data['path'].startswith('zvol/'):
             zvolname = zvol_path_to_name(os.path.join('/dev', data['path']))
-            await self.middleware.call('zfs.dataset.update', zvolname, {'properties': {'volthreading': {'value': 'off'}}})
+            await self.middleware.call(
+                'zfs.dataset.update',
+                zvolname,
+                {
+                    'properties': {
+                        'volthreading': {'value': 'off'},
+                        'readonly': {'value': 'on' if data['ro'] else 'off'}
+                    }
+                }
+            )
 
         data['id'] = await self.middleware.call(
             'datastore.insert', self._config.datastore, {**data, 'vendor': 'TrueNAS'},
@@ -145,15 +154,26 @@ async def do_update(self, audit_callback, id_, data):
         new.update(data)
 
         await self.middleware.call('iscsi.extent.validate', new)
-        await self.clean(
-            new, 'iscsi_extent_update', verrors, old=old
-        )
+        await self.clean(new, 'iscsi_extent_update', verrors, old=old)
         verrors.check()
 
         await self.middleware.call('iscsi.extent.save', new, 'iscsi_extent_create', verrors, old)
         verrors.check()
         new.pop(self.locked_field)
 
+        zvolpath = new.get('path')
+        if zvolpath is not None and zvolpath.startswith('zvol/'):
+            zvolname = zvol_path_to_name(os.path.join('/dev', zvolpath))
+            await self.middleware.call(
+                'zfs.dataset.update',
+                zvolname,
+                {
+                    'properties': {
+                        'readonly': {'value': 'on' if new['ro'] else 'off'}
+                    }
+                }
+            )
+
         await self.middleware.call(
             'datastore.update',
             self._config.datastore,
@@ -164,6 +184,9 @@ async def do_update(self, audit_callback, id_, data):
 
         await self._service_change('iscsitarget', 'reload')
 
+        # scstadmin can have issues when modifying an existing extent, re-run
+        await self._service_change('iscsitarget', 'reload')
+
         return await self.get_instance(id_)
 
     @accepts(
@@ -219,8 +242,8 @@ async def do_delete(self, audit_callback, id_, remove, force):
                 await self.middleware.call('iscsi.alua.wait_for_alua_settled')
 
     @private
-    def validate(self, data):
-        data['serial'] = self.extent_serial(data['serial'])
+    async def validate(self, data):
+        data['serial'] = await self.extent_serial(data['serial'])
         data['naa'] = self.extent_naa(data.get('naa'))
 
     @private
@@ -386,10 +409,10 @@ def clean_size(self, data, schema_name, verrors):
         return data
 
     @private
-    def extent_serial(self, serial):
+    async def extent_serial(self, serial):
         if serial in [None, '']:
             used_serials = [i['serial'] for i in (
-                self.middleware.call_sync('iscsi.extent.query', [], {'select': ['serial']})
+                await self.middleware.call('iscsi.extent.query', [], {'select': ['serial']})
             )]
             tries = 5
             for i in range(tries):
diff --git a/src/middlewared/middlewared/plugins/iscsi_/global_linux.py b/src/middlewared/middlewared/plugins/iscsi_/global_linux.py
index 4e889eed25e0..07653f461951 100644
--- a/src/middlewared/middlewared/plugins/iscsi_/global_linux.py
+++ b/src/middlewared/middlewared/plugins/iscsi_/global_linux.py
@@ -3,6 +3,7 @@
 
 from middlewared.schema import Bool, Dict, Int, Str
 from middlewared.service import filterable, filterable_returns, private, Service
+from middlewared.service_exception import MatchNotFound
 from middlewared.utils import filter_list, run
 
 
@@ -94,6 +95,20 @@ def sessions(self, filters, options):
                 sessions.append(session_dict)
         return filter_list(sessions, filters, options)
 
+    @private
+    def resync_readonly_property_for_zvol(self, id_, read_only_value):
+        try:
+            extent = self.middleware.call_sync(
+                'iscsi.extent.query',
+                [['enabled', '=', True], ['path', '=', f'zvol/{id_}']],
+                {'get': True}
+            )
+            ro = True if read_only_value.lower() == 'on' else False
+            if extent['ro'] != ro:
+                self.middleware.call_sync('iscsi.extent.update', extent['id'], {'ro': ro})
+        except MatchNotFound:
+            return
+
     @private
     def resync_lun_size_for_zvol(self, id_):
         if not self.middleware.call_sync('service.started', 'iscsitarget'):
diff --git a/src/middlewared/middlewared/plugins/pool_/dataset.py b/src/middlewared/middlewared/plugins/pool_/dataset.py
index 416a58a74ead..a853914bbfce 100644
--- a/src/middlewared/middlewared/plugins/pool_/dataset.py
+++ b/src/middlewared/middlewared/plugins/pool_/dataset.py
@@ -939,10 +939,18 @@ async def do_update(self, audit_callback, id_, data):
             verrors.add_child('pool_dataset_update', self.__handle_zfs_set_property_error(e, properties_definitions))
             raise verrors
 
-        if data['type'] == 'VOLUME' and 'volsize' in data and data['volsize'] > dataset[0]['volsize']['parsed']:
-            # means the zvol size has increased so we need to check if this zvol is shared via SCST (iscsi)
-            # and if it is, resync it so the connected initiators can see the new size of the zvol
-            await self.middleware.call('iscsi.global.resync_lun_size_for_zvol', id_)
+        if data['type'] == 'VOLUME':
+            if 'volsize' in data and data['volsize'] > dataset[0]['volsize']['parsed']:
+                # means the zvol size has increased so we need to check if this zvol is shared via SCST (iscsi)
+                # and if it is, resync it so the connected initiators can see the new size of the zvol
+                await self.middleware.call('iscsi.global.resync_lun_size_for_zvol', id_)
+
+            if 'readonly' in data:
+                # depending on the iscsi client connected to us, if someone marks a zvol
+                # as R/O (or R/W), we need to be sure and update the associated extent so
+                # that we don't get into a scenario where the iscsi extent is R/W but the
+                # underlying zvol is R/O. Windows clients seem to not handle this very well.
+                await self.middleware.call('iscsi.global.resync_readonly_property_for_zvol', id_, data['readonly'])
 
         updated_ds = await self.get_instance(id_)
         self.middleware.send_event('pool.dataset.query', 'CHANGED', id=id_, fields=updated_ds)
diff --git a/tests/api2/test_261_iscsi_cmd.py b/tests/api2/test_261_iscsi_cmd.py
index 045a927fb9d7..b6316fb76551 100644
--- a/tests/api2/test_261_iscsi_cmd.py
+++ b/tests/api2/test_261_iscsi_cmd.py
@@ -21,7 +21,7 @@
 from middlewared.test.integration.assets.pool import dataset, snapshot
 from middlewared.test.integration.utils import call, ssh
 from middlewared.test.integration.utils.client import truenas_server
-from pyscsi.pyscsi.scsi_sense import sense_ascq_dict
+from pyscsi.pyscsi.scsi_sense import sense_ascq_dict, sense_key_dict
 from pytest_dependency import depends
 
 from auto_config import ha, hostname, isns_ip, password, pool_name, user
@@ -224,8 +224,16 @@ def get_client_count():
     return call('iscsi.global.client_count')
 
 
+def get_zvol_property(zvolid, property_name):
+    return call('zfs.dataset.query', [['id', '=', zvolid]], {'get': True})['properties'][property_name]['value']
+
+
 def get_volthreading(zvolid):
-    return call('zfs.dataset.query', [['id', '=', zvolid]], {'get': True})['properties']['volthreading']['value']
+    return get_zvol_property(zvolid, 'volthreading')
+
+
+def get_readonly(zvolid):
+    return get_zvol_property(zvolid, 'readonly')
 
 
 def verify_client_count(count, retries=10):
@@ -407,6 +415,14 @@ def expect_check_condition(s, text=None, check_type=CheckType.CHECK_CONDITION):
             s.testunitready()
 
 
+@contextlib.contextmanager
+def raises_check_condition(sense_key, ascq):
+    with pytest.raises(Exception) as excinfo:
+        yield
+    e = excinfo.value
+    assert f"Check Condition: {sense_key_dict[sense_key]}(0x{sense_key:02X}) ASC+Q:{sense_ascq_dict[ascq]}(0x{ascq:04X})" == str(e)
+
+
 def _verify_inquiry(s):
     """
     Verify that the supplied SCSI has the expected INQUIRY response.
@@ -2722,6 +2738,95 @@ def test_35_delete_extent_no_dataset(extent_type):
                     with zvol_extent(zvol, extent_name='zvolextent1'):
                         ssh(DESTROY_CMD)
 
+def test_36_target_readonly_extent():
+    """Validate a target that is made RO - either by modifying the extent
+    setting, or the underlying ZVOL - behaves correctly."""
+    name1 = f"{target_name}x1"
+    iqn = f'{basename}:{name1}'
+
+    zeros = bytearray(512)
+    deadbeef = bytearray.fromhex('deadbeef') * 128
+
+    def lba_data(flipped):
+        if flipped:
+            return deadbeef, zeros
+        else:
+            return zeros, deadbeef
+
+    def write_lbas(s, flipped=False):
+        lba0, lba1 = lba_data(flipped)
+        s.write16(0, 1, lba0)
+        s.write16(1, 1, lba1)
+
+    def read_lbas(s, flipped=False):
+        lba0, lba1 = lba_data(flipped)
+        r = s.read16(0, 1)
+        assert r.datain == lba0, r.datain
+        r = s.read16(1, 1)
+        assert r.datain == lba1, r.datain
+
+    def check_readonly_state(zvolid, extentid, readonly):
+        if readonly:
+            assert get_readonly(zvolid) == 'on'
+            assert call('iscsi.extent.get_instance', extentid)['ro'] is True
+        else:
+            assert get_readonly(zvolid) == 'off'
+            assert call('iscsi.extent.get_instance', extentid)['ro'] is False
+
+    with initiator_portal() as config:
+        with configured_target(config, name1, 'VOLUME') as target_config:
+            with iscsi_scsi_connection(truenas_server.ip, iqn) as s:
+                zvolid = target_config['dataset']
+                extentid = target_config['extent']['id']
+
+                # Ensure that we can read and write
+                write_lbas(s)
+                read_lbas(s)
+                check_readonly_state(zvolid, extentid, False)
+
+                # Set RO by updating the extent
+                call('iscsi.extent.update', extentid, {'ro': True})
+                expect_check_condition(s, sense_ascq_dict[0x2900])  # "POWER ON, RESET, OR BUS DEVICE RESET OCCURRED"
+                check_readonly_state(zvolid, extentid, True)
+
+                # Ensure that we can only READ
+                read_lbas(s)
+                # Write => Check Condition Sense key = 7 for Data Protect, ASCQ == 0
+                with raises_check_condition(7, 0):
+                    write_lbas(s, True)
+                read_lbas(s)
+
+                # Set RW by updating the extent
+                call('iscsi.extent.update', extentid, {'ro': False})
+                expect_check_condition(s, sense_ascq_dict[0x2900])  # "POWER ON, RESET, OR BUS DEVICE RESET OCCURRED"
+                check_readonly_state(zvolid, extentid, False)
+
+                # Ensure that we can read and write
+                read_lbas(s)
+                write_lbas(s, True)
+                read_lbas(s, True)
+
+                # Set RO by updating the ZVOL
+                call('pool.dataset.update', zvolid, {'readonly': 'ON'})
+                expect_check_condition(s, sense_ascq_dict[0x2900])  # "POWER ON, RESET, OR BUS DEVICE RESET OCCURRED"
+                check_readonly_state(zvolid, extentid, True)
+
+                # Ensure that we can only READ
+                read_lbas(s, True)
+                with raises_check_condition(7, 0):
+                    write_lbas(s)
+                read_lbas(s, True)
+
+                # Set RW by updating the ZVOL
+                call('pool.dataset.update', zvolid, {'readonly': 'OFF'})
+                expect_check_condition(s, sense_ascq_dict[0x2900])  # "POWER ON, RESET, OR BUS DEVICE RESET OCCURRED"
+                check_readonly_state(zvolid, extentid, False)
+
+                # Ensure that we can read and write
+                read_lbas(s, True)
+                write_lbas(s)
+                read_lbas(s)
+
 
 def test_99_teardown(request):
     # Disable iSCSI service