diff --git a/releasenotes/notes/add-volume-detach-libs-2cbb3ca924aed0ac.yaml b/releasenotes/notes/add-volume-detach-libs-2cbb3ca924aed0ac.yaml
new file mode 100644
index 0000000..30127b3
--- /dev/null
+++ b/releasenotes/notes/add-volume-detach-libs-2cbb3ca924aed0ac.yaml
@@ -0,0 +1,5 @@
+---
+features:
+  - |
+    Add delete_attachment to the v3 AttachmentsClient and terminate_connection
+    to the v3 VolumesClient.
diff --git a/releasenotes/source/index.rst b/releasenotes/source/index.rst
index 882413f..4c1edd5 100644
--- a/releasenotes/source/index.rst
+++ b/releasenotes/source/index.rst
@@ -6,6 +6,7 @@
    :maxdepth: 1
 
    unreleased
+   v34.2.0
    v34.0.0
    v33.0.0
    v32.0.0
diff --git a/releasenotes/source/v34.2.0.rst b/releasenotes/source/v34.2.0.rst
new file mode 100644
index 0000000..386cf71
--- /dev/null
+++ b/releasenotes/source/v34.2.0.rst
@@ -0,0 +1,6 @@
+=====================
+v34.2.0 Release Notes
+=====================
+
+.. release-notes:: 34.2.0 Release Notes
+   :version: 34.2.0
diff --git a/tempest/api/image/base.py b/tempest/api/image/base.py
index 23e7fd8..11a1e6c 100644
--- a/tempest/api/image/base.py
+++ b/tempest/api/image/base.py
@@ -13,6 +13,7 @@
 #    under the License.
 
 import io
+import time
 
 from tempest.common import image as common_image
 from tempest import config
@@ -22,6 +23,7 @@
 import tempest.test
 
 CONF = config.CONF
+BAD_REQUEST_RETRIES = 3
 
 
 class BaseImageTest(tempest.test.BaseTestCase):
@@ -159,6 +161,82 @@
             pass
         return stores
 
+    def _update_image_with_retries(self, image, patch):
+        # NOTE(danms): If glance was unable to fetch the remote image via
+        # HTTP, it will return BadRequest. Because this can be transient in
+        # CI, we try this a few times before we agree that it has failed
+        # for a reason worthy of failing the test.
+        for i in range(BAD_REQUEST_RETRIES):
+            try:
+                self.client.update_image(image, patch)
+                break
+            except exceptions.BadRequest:
+                if i + 1 == BAD_REQUEST_RETRIES:
+                    raise
+                else:
+                    time.sleep(1)
+
+    def check_set_location(self):
+        image = self.client.create_image(container_format='bare',
+                                         disk_format='raw')
+
+        # Locations should be empty when there is no data
+        self.assertEqual('queued', image['status'])
+        self.assertEqual([], image['locations'])
+
+        # Add a new location
+        new_loc = {'metadata': {'foo': 'bar'},
+                   'url': CONF.image.http_image}
+        self._update_image_with_retries(image['id'], [
+            dict(add='/locations/-', value=new_loc)])
+
+        # The image should now be active, with one location that looks
+        # like we expect
+        image = self.client.show_image(image['id'])
+        self.assertEqual(1, len(image['locations']),
+                         'Image should have one location but has %i' % (
+                         len(image['locations'])))
+        self.assertEqual(new_loc['url'], image['locations'][0]['url'])
+        self.assertEqual('bar', image['locations'][0]['metadata'].get('foo'))
+        if 'direct_url' in image:
+            self.assertEqual(image['direct_url'], image['locations'][0]['url'])
+
+        # If we added the location directly, the image goes straight
+        # to active and no hashing is done
+        self.assertEqual('active', image['status'])
+        self.assertIsNone(None, image['os_hash_algo'])
+        self.assertIsNone(None, image['os_hash_value'])
+
+        return image
+
+    def check_set_multiple_locations(self):
+        image = self.check_set_location()
+
+        new_loc = {'metadata': {'speed': '88mph'},
+                   'url': '%s#new' % CONF.image.http_image}
+        self._update_image_with_retries(image['id'],
+                                        [dict(add='/locations/-',
+                                              value=new_loc)])
+
+        # The image should now have two locations and the last one
+        # (locations are ordered) should have the new URL.
+        image = self.client.show_image(image['id'])
+        self.assertEqual(2, len(image['locations']),
+                         'Image should have two locations but has %i' % (
+                         len(image['locations'])))
+        self.assertEqual(new_loc['url'], image['locations'][1]['url'])
+
+        # The image should still be active and still have no hashes
+        self.assertEqual('active', image['status'])
+        self.assertIsNone(None, image['os_hash_algo'])
+        self.assertIsNone(None, image['os_hash_value'])
+
+        # The direct_url should still match the first location
+        if 'direct_url' in image:
+            self.assertEqual(image['direct_url'], image['locations'][0]['url'])
+
+        return image
+
 
 class BaseV2MemberImageTest(BaseV2ImageTest):
 
diff --git a/tempest/api/image/v2/admin/test_images.py b/tempest/api/image/v2/admin/test_images.py
index 733c778..a77b2f2 100644
--- a/tempest/api/image/v2/admin/test_images.py
+++ b/tempest/api/image/v2/admin/test_images.py
@@ -20,6 +20,7 @@
 from tempest import config
 from tempest.lib.common.utils import data_utils
 from tempest.lib import decorators
+from tempest.lib import exceptions as lib_exc
 
 CONF = config.CONF
 
@@ -120,3 +121,40 @@
         self.assertEqual(0, len(failed_stores),
                          "Failed to copy the following stores: %s" %
                          str(failed_stores))
+
+
+class ImageLocationsAdminTest(base.BaseV2ImageAdminTest):
+
+    @classmethod
+    def skip_checks(cls):
+        super(ImageLocationsAdminTest, cls).skip_checks()
+        if not CONF.image_feature_enabled.manage_locations:
+            skip_msg = (
+                "%s skipped as show_multiple_locations is not available" % (
+                    cls.__name__))
+            raise cls.skipException(skip_msg)
+
+    @decorators.idempotent_id('8a648de4-b745-4c28-a7b5-20de1c3da4d2')
+    def test_delete_locations(self):
+        image = self.check_set_multiple_locations()
+        expected_remaining_loc = image['locations'][1]
+
+        self.admin_client.update_image(image['id'], [
+            dict(remove='/locations/0')])
+
+        # The image should now have only the one location we did not delete
+        image = self.client.show_image(image['id'])
+        self.assertEqual(1, len(image['locations']),
+                         'Image should have one location but has %i' % (
+                         len(image['locations'])))
+        self.assertEqual(expected_remaining_loc['url'],
+                         image['locations'][0]['url'])
+
+        # The direct_url should now be the last remaining location
+        if 'direct_url' in image:
+            self.assertEqual(image['direct_url'], image['locations'][0]['url'])
+
+        # Removing the last location should be disallowed
+        self.assertRaises(lib_exc.Forbidden,
+                          self.admin_client.update_image, image['id'], [
+                              dict(remove='/locations/0')])
diff --git a/tempest/api/image/v2/test_images.py b/tempest/api/image/v2/test_images.py
index b723977..fecd5a7 100644
--- a/tempest/api/image/v2/test_images.py
+++ b/tempest/api/image/v2/test_images.py
@@ -16,7 +16,6 @@
 
 import io
 import random
-import time
 
 from oslo_log import log as logging
 from tempest.api.image import base
@@ -28,7 +27,6 @@
 
 CONF = config.CONF
 LOG = logging.getLogger(__name__)
-BAD_REQUEST_RETRIES = 3
 
 
 class ImportImagesTest(base.BaseV2ImageTest):
@@ -808,89 +806,13 @@
 
         return image
 
-    def _check_set_location(self):
-        image = self.client.create_image(container_format='bare',
-                                         disk_format='raw')
-
-        # Locations should be empty when there is no data
-        self.assertEqual('queued', image['status'])
-        self.assertEqual([], image['locations'])
-
-        # Add a new location
-        new_loc = {'metadata': {'foo': 'bar'},
-                   'url': CONF.image.http_image}
-        self._update_image_with_retries(image['id'], [
-            dict(add='/locations/-', value=new_loc)])
-
-        # The image should now be active, with one location that looks
-        # like we expect
-        image = self.client.show_image(image['id'])
-        self.assertEqual(1, len(image['locations']),
-                         'Image should have one location but has %i' % (
-                         len(image['locations'])))
-        self.assertEqual(new_loc['url'], image['locations'][0]['url'])
-        self.assertEqual('bar', image['locations'][0]['metadata'].get('foo'))
-        if 'direct_url' in image:
-            self.assertEqual(image['direct_url'], image['locations'][0]['url'])
-
-        # If we added the location directly, the image goes straight
-        # to active and no hashing is done
-        self.assertEqual('active', image['status'])
-        self.assertIsNone(None, image['os_hash_algo'])
-        self.assertIsNone(None, image['os_hash_value'])
-
-        return image
-
     @decorators.idempotent_id('37599b8a-d5c0-4590-aee5-73878502be15')
     def test_set_location(self):
-        self._check_set_location()
-
-    def _update_image_with_retries(self, image, patch):
-        # NOTE(danms): If glance was unable to fetch the remote image via
-        # HTTP, it will return BadRequest. Because this can be transient in
-        # CI, we try this a few times before we agree that it has failed
-        # for a reason worthy of failing the test.
-        for i in range(BAD_REQUEST_RETRIES):
-            try:
-                self.client.update_image(image, patch)
-                break
-            except lib_exc.BadRequest:
-                if i + 1 == BAD_REQUEST_RETRIES:
-                    raise
-                else:
-                    time.sleep(1)
-
-    def _check_set_multiple_locations(self):
-        image = self._check_set_location()
-
-        new_loc = {'metadata': {'speed': '88mph'},
-                   'url': '%s#new' % CONF.image.http_image}
-        self._update_image_with_retries(image['id'],
-                                        [dict(add='/locations/-',
-                                              value=new_loc)])
-
-        # The image should now have two locations and the last one
-        # (locations are ordered) should have the new URL.
-        image = self.client.show_image(image['id'])
-        self.assertEqual(2, len(image['locations']),
-                         'Image should have two locations but has %i' % (
-                         len(image['locations'])))
-        self.assertEqual(new_loc['url'], image['locations'][1]['url'])
-
-        # The image should still be active and still have no hashes
-        self.assertEqual('active', image['status'])
-        self.assertIsNone(None, image['os_hash_algo'])
-        self.assertIsNone(None, image['os_hash_value'])
-
-        # The direct_url should still match the first location
-        if 'direct_url' in image:
-            self.assertEqual(image['direct_url'], image['locations'][0]['url'])
-
-        return image
+        self.check_set_location()
 
     @decorators.idempotent_id('bf6e0009-c039-4884-b498-db074caadb10')
     def test_replace_location(self):
-        image = self._check_set_multiple_locations()
+        image = self.check_set_multiple_locations()
         original_locs = image['locations']
 
         # Replacing with the exact thing should work
@@ -927,31 +849,6 @@
                          len(image['locations'])))
         self.assertEqual(original_locs, image['locations'])
 
-    @decorators.idempotent_id('8a648de4-b745-4c28-a7b5-20de1c3da4d2')
-    def test_delete_locations(self):
-        image = self._check_set_multiple_locations()
-        expected_remaining_loc = image['locations'][1]
-
-        self.client.update_image(image['id'], [
-            dict(remove='/locations/0')])
-
-        # The image should now have only the one location we did not delete
-        image = self.client.show_image(image['id'])
-        self.assertEqual(1, len(image['locations']),
-                         'Image should have one location but has %i' % (
-                         len(image['locations'])))
-        self.assertEqual(expected_remaining_loc['url'],
-                         image['locations'][0]['url'])
-
-        # The direct_url should now be the last remaining location
-        if 'direct_url' in image:
-            self.assertEqual(image['direct_url'], image['locations'][0]['url'])
-
-        # Removing the last location should be disallowed
-        self.assertRaises(lib_exc.Forbidden,
-                          self.client.update_image, image['id'], [
-                              dict(remove='/locations/0')])
-
     @decorators.idempotent_id('a9a20396-8399-4b36-909d-564949be098f')
     def test_set_location_bad_scheme(self):
         image = self.client.create_image(container_format='bare',
diff --git a/tempest/api/volume/admin/test_volumes_actions.py b/tempest/api/volume/admin/test_volumes_actions.py
index ecddfba..b6e9f32 100644
--- a/tempest/api/volume/admin/test_volumes_actions.py
+++ b/tempest/api/volume/admin/test_volumes_actions.py
@@ -83,7 +83,7 @@
         server_id = self.create_server()['id']
         volume_id = self.create_volume()['id']
 
-        # Attach volume
+        # Request Cinder to map & export volume (it's not attached to instance)
         self.volumes_client.attach_volume(
             volume_id,
             instance_uuid=server_id,
@@ -101,7 +101,9 @@
         waiters.wait_for_volume_resource_status(self.volumes_client,
                                                 volume_id, 'error')
 
-        # Force detach volume
+        # The force detach volume calls works because the volume is not really
+        # connected to the instance (it is safe), otherwise it would be
+        # rejected for security reasons (bug #2004555).
         self.admin_volume_client.force_detach_volume(
             volume_id, connector=None,
             attachment_id=attachment['attachment_id'])
diff --git a/tempest/lib/api_schema/response/volume/volumes.py b/tempest/lib/api_schema/response/volume/volumes.py
index 4f44526..900e5ef 100644
--- a/tempest/lib/api_schema/response/volume/volumes.py
+++ b/tempest/lib/api_schema/response/volume/volumes.py
@@ -295,6 +295,7 @@
 attach_volume = {'status_code': [202]}
 set_bootable_volume = {'status_code': [200]}
 detach_volume = {'status_code': [202]}
+terminate_connection = {'status_code': [202]}
 reserve_volume = {'status_code': [202]}
 unreserve_volume = {'status_code': [202]}
 extend_volume = {'status_code': [202]}
diff --git a/tempest/lib/services/volume/v3/attachments_client.py b/tempest/lib/services/volume/v3/attachments_client.py
index 5e448f7..303341e 100644
--- a/tempest/lib/services/volume/v3/attachments_client.py
+++ b/tempest/lib/services/volume/v3/attachments_client.py
@@ -26,3 +26,10 @@
         body = json.loads(body)
         self.expected_success(200, resp.status)
         return rest_client.ResponseBody(resp, body)
+
+    def delete_attachment(self, attachment_id):
+        """Delete volume attachment."""
+        url = "attachments/%s" % (attachment_id)
+        resp, body = self.delete(url)
+        self.expected_success(200, resp.status)
+        return rest_client.ResponseBody(resp, body)
diff --git a/tempest/lib/services/volume/v3/volumes_client.py b/tempest/lib/services/volume/v3/volumes_client.py
index ad8bd71..c6f8973 100644
--- a/tempest/lib/services/volume/v3/volumes_client.py
+++ b/tempest/lib/services/volume/v3/volumes_client.py
@@ -205,14 +205,23 @@
         self.validate_response(schema.set_bootable_volume, resp, body)
         return rest_client.ResponseBody(resp, body)
 
-    def detach_volume(self, volume_id):
+    def detach_volume(self, volume_id, **kwargs):
         """Detaches a volume from an instance."""
-        post_body = json.dumps({'os-detach': {}})
+        post_body = json.dumps({'os-detach': kwargs})
         url = 'volumes/%s/action' % (volume_id)
         resp, body = self.post(url, post_body)
         self.validate_response(schema.detach_volume, resp, body)
         return rest_client.ResponseBody(resp, body)
 
+    def terminate_connection(self, volume_id, connector):
+        """Detaches a volume from an instance using terminate_connection."""
+        post_body = json.dumps(
+            {'os-terminate_connection': {'connector': connector}})
+        url = 'volumes/%s/action' % (volume_id)
+        resp, body = self.post(url, post_body)
+        self.validate_response(schema.terminate_connection, resp, body)
+        return rest_client.ResponseBody(resp, body)
+
     def reserve_volume(self, volume_id):
         """Reserves a volume."""
         post_body = json.dumps({'os-reserve': {}})
diff --git a/tempest/scenario/test_server_volume_attachment.py b/tempest/scenario/test_server_volume_attachment.py
new file mode 100644
index 0000000..cc8cf00
--- /dev/null
+++ b/tempest/scenario/test_server_volume_attachment.py
@@ -0,0 +1,189 @@
+# Copyright 2023 Red Hat
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from unittest import mock
+
+from tempest.common import utils
+from tempest.common import waiters
+from tempest import config
+from tempest.lib import decorators
+from tempest.lib import exceptions
+from tempest.scenario import manager
+
+CONF = config.CONF
+
+
+class BaseAttachmentTest(manager.ScenarioTest):
+    @classmethod
+    def setup_clients(cls):
+        super().setup_clients()
+        cls.attachments_client = cls.os_primary.attachments_client_latest
+        cls.admin_volume_client = cls.os_admin.volumes_client_latest
+
+    def _call_with_fake_service_token(self, valid_token,
+                                      client, method_name, *args, **kwargs):
+        """Call client method with non-service service token
+
+        Add a service token header that can be a valid normal user token (which
+        won't have the service role) or an invalid token altogether.
+        """
+        original_raw_request = client.raw_request
+
+        def raw_request(url, method, headers=None, body=None, chunked=False,
+                        log_req_body=None):
+            token = headers['X-Auth-Token']
+            if not valid_token:
+                token = token[:-1] + ('a' if token[-1] != 'a' else 'b')
+            headers['X-Service-Token'] = token
+            return original_raw_request(url, method, headers=headers,
+                                        body=body, chunked=chunked,
+                                        log_req_body=log_req_body)
+
+        client_method = getattr(client, method_name)
+        with mock.patch.object(client, 'raw_request', raw_request):
+            return client_method(*args, **kwargs)
+
+
+class TestServerVolumeAttachmentScenario(BaseAttachmentTest):
+
+    """Test server attachment behaviors
+
+    This tests that volume attachments to servers may not be removed directly
+    and are only allowed through the compute service (bug #2004555).
+    """
+
+    @decorators.attr(type='slow')
+    @decorators.idempotent_id('be615530-f105-437a-8afe-ce998c9535d9')
+    @utils.services('compute', 'volume', 'image', 'network')
+    def test_server_detach_rules(self):
+        """Test that various methods of detaching a volume honors the rules"""
+        server = self.create_server(wait_until='SSHABLE')
+        servers = self.servers_client.list_servers()['servers']
+        self.assertIn(server['id'], [x['id'] for x in servers])
+
+        volume = self.create_volume()
+
+        volume = self.nova_volume_attach(server, volume)
+        self.addCleanup(self.nova_volume_detach, server, volume)
+        att_id = volume['attachments'][0]['attachment_id']
+
+        # Test user call to detach volume is rejected
+        self.assertRaises((exceptions.Forbidden, exceptions.Conflict),
+                          self.volumes_client.detach_volume, volume['id'])
+
+        # Test user call to terminate connection is rejected
+        self.assertRaises((exceptions.Forbidden, exceptions.Conflict),
+                          self.volumes_client.terminate_connection,
+                          volume['id'], connector={})
+
+        # Test faking of service token on call to detach, force detach,
+        # terminate_connection
+        for valid_token in (True, False):
+            valid_exceptions = [exceptions.Forbidden, exceptions.Conflict]
+            if not valid_token:
+                valid_exceptions.append(exceptions.Unauthorized)
+            self.assertRaises(
+                tuple(valid_exceptions),
+                self._call_with_fake_service_token,
+                valid_token,
+                self.volumes_client,
+                'detach_volume',
+                volume['id'])
+            self.assertRaises(
+                tuple(valid_exceptions),
+                self._call_with_fake_service_token,
+                valid_token,
+                self.volumes_client,
+                'terminate_connection',
+                volume['id'], connector={})
+
+        # Reset volume's status to error
+        self.admin_volume_client.reset_volume_status(volume['id'],
+                                                     status='error')
+        waiters.wait_for_volume_resource_status(self.volumes_client,
+                                                volume['id'], 'error')
+
+        # For the cleanup, we need to reset the volume status to in-use before
+        # the other cleanup steps try to detach it.
+        self.addCleanup(waiters.wait_for_volume_resource_status,
+                        self.volumes_client, volume['id'], 'in-use')
+        self.addCleanup(self.admin_volume_client.reset_volume_status,
+                        volume['id'], status='in-use')
+
+        # Test user call to force detach volume is rejected
+        self.assertRaises(
+            (exceptions.Forbidden, exceptions.Conflict),
+            self.admin_volume_client.force_detach_volume,
+            volume['id'], connector=None,
+            attachment_id=att_id)
+
+        # Test trying to override detach with force and service token
+        for valid_token in (True, False):
+            valid_exceptions = [exceptions.Forbidden, exceptions.Conflict]
+            if not valid_token:
+                valid_exceptions.append(exceptions.Unauthorized)
+            self.assertRaises(
+                tuple(valid_exceptions),
+                self._call_with_fake_service_token,
+                valid_token,
+                self.admin_volume_client,
+                'force_detach_volume',
+                volume['id'], connector=None, attachment_id=att_id)
+
+        # Test user call to detach with mismatch is rejected
+        volume2 = self.create_volume()
+        volume2 = self.nova_volume_attach(server, volume2)
+        att_id2 = volume2['attachments'][0]['attachment_id']
+        self.assertRaises(
+            (exceptions.Forbidden, exceptions.BadRequest),
+            self.volumes_client.detach_volume,
+            volume['id'], attachment_id=att_id2)
+
+
+class TestServerVolumeAttachScenarioOldVersion(BaseAttachmentTest):
+    volume_min_microversion = '3.27'
+    volume_max_microversion = 'latest'
+
+    @decorators.attr(type='slow')
+    @decorators.idempotent_id('6f4d2144-99f4-495c-8b0b-c6a537971418')
+    @utils.services('compute', 'volume', 'image', 'network')
+    def test_old_versions_reject(self):
+        server = self.create_server(wait_until='SSHABLE')
+        servers = self.servers_client.list_servers()['servers']
+        self.assertIn(server['id'], [x['id'] for x in servers])
+
+        volume = self.create_volume()
+
+        volume = self.nova_volume_attach(server, volume)
+        self.addCleanup(self.nova_volume_detach, server, volume)
+        att_id = volume['attachments'][0]['attachment_id']
+
+        for valid_token in (True, False):
+            valid_exceptions = [exceptions.Forbidden,
+                                exceptions.Conflict]
+            if not valid_token:
+                valid_exceptions.append(exceptions.Unauthorized)
+            self.assertRaises(
+                tuple(valid_exceptions),
+                self._call_with_fake_service_token,
+                valid_token,
+                self.attachments_client,
+                'delete_attachment',
+                att_id)
+
+        self.assertRaises(
+            (exceptions.Forbidden, exceptions.Conflict),
+            self.attachments_client.delete_attachment,
+            att_id)