Implement sharing of zones

Author: Igor Malinovskiy <u.glide@gmail.com>
Co-Authored-By: Sergey Drozdov <sergey.drozdov.dev@gmail.com, sergey.drozdov93@thehutgroup.com>
Co-Authored-By: Michael Johnson <johnsomor@gmail.com>

Change-Id: Ibd780f3c695a95be00ff97d7736d5a0bebea79b9
Closes-Bug: #1714088
Depends-On: https://review.opendev.org/c/openstack/designate-tempest-plugin/+/872069

api-ref/source/dns-api-v2-index.rst

@@ -10,6 +10,7 @@
 .. include:: dns-api-v2-zone-import.inc
 .. include:: dns-api-v2-zone-export.inc
 .. include:: dns-api-v2-zone-tasks.inc
+.. include:: dns-api-v2-shared-zones.inc
 .. include:: dns-api-v2-zone-ownership-transfer-request.inc
 .. include:: dns-api-v2-zone-ownership-transfer-accept.inc
 .. include:: dns-api-v2-recordset.inc

api-ref/source/dns-api-v2-shared-zones.inc

@@ -0,0 +1,215 @@
+============
+Shared Zones
+============
+
+Shared zones operations.
+
+
+Show a Zone Share
+=================
+
+.. rest_method::  GET /v2/zones/{zone_id}/shares/{zone_share_id}
+
+Show a single zone share.
+
+**New in version 2.1**
+
+.. rest_status_code:: success status.yaml
+
+   - 200
+
+.. rest_status_code:: error status.yaml
+
+   - 400
+   - 401
+   - 403
+   - 404
+   - 405
+   - 500
+   - 503
+
+Request
+-------
+
+.. rest_parameters:: parameters.yaml
+
+   - x-auth-token: x-auth-token
+   - x-auth-all-projects: x-auth-all-projects
+   - x-auth-sudo-project-id: x-auth-sudo-project-id
+   - zone_id: path_zone_id
+   - zone_share_id: path_zone_share_id
+
+Response Parameters
+-------------------
+
+.. rest_parameters:: parameters.yaml
+
+   - x-openstack-request-id: x-openstack-request-id
+   - id: id
+   - zone_id: shared_zone_id
+   - project_id: project_id
+   - target_project_id: target_project_id
+   - created_at: created_at
+   - updated_at: updated_at
+   - links: links
+
+Response Example
+----------------
+
+.. literalinclude:: samples/zones/share-zone-response.json
+
+Get All Shared Zones
+====================
+
+.. rest_method::  GET /v2/zones/{zone_id}/shares
+
+List all zone shares.
+
+**New in version 2.1**
+
+.. rest_status_code:: success status.yaml
+
+   - 200
+
+.. rest_status_code:: error status.yaml
+
+   - 400
+   - 401
+   - 403
+   - 404
+   - 405
+   - 500
+   - 503
+
+
+Request
+-------
+
+.. rest_parameters:: parameters.yaml
+
+   - x-auth-token: x-auth-token
+   - x-auth-all-projects: x-auth-all-projects
+   - x-auth-sudo-project-id: x-auth-sudo-project-id
+   - zone_id: path_zone_id
+   - target_project_id: target_project_id_filter
+
+Response Parameters
+-------------------
+
+.. rest_parameters:: parameters.yaml
+
+   - x-openstack-request-id: x-openstack-request-id
+   - id: id
+   - zone_id: shared_zone_id
+   - project_id: project_id
+   - target_project_id: target_project_id
+   - created_at: created_at
+   - updated_at: updated_at
+   - links: links
+
+Response Example
+----------------
+
+.. literalinclude:: samples/zones/list-share-zone-response.json
+
+
+Create Shared Zone
+==================
+
+.. rest_method::  POST /v2/zones/{zone_id}/shares
+
+Share a zone with another project.
+
+**New in version 2.1**
+
+.. rest_status_code:: success status.yaml
+
+   - 201
+
+.. rest_status_code:: error status.yaml
+
+   - 400
+   - 401
+   - 403
+   - 404
+   - 405
+   - 409
+   - 500
+   - 503
+
+Request
+-------
+
+.. rest_parameters:: parameters.yaml
+
+   - x-auth-token: x-auth-token
+   - x-auth-all-projects: x-auth-all-projects
+   - x-auth-sudo-project-id: x-auth-sudo-project-id
+   - zone_id: path_zone_id
+   - target_project_id: target_project_id
+
+Request Example
+---------------
+
+.. literalinclude:: samples/zones/share-zone-request.json
+
+Response Parameters
+-------------------
+
+.. rest_parameters:: parameters.yaml
+
+   - x-openstack-request-id: x-openstack-request-id
+   - id: id
+   - zone_id: shared_zone_id
+   - project_id: project_id
+   - target_project_id: target_project_id
+   - created_at: created_at
+   - updated_at: updated_at
+   - links: links
+
+Response Example
+----------------
+
+.. literalinclude:: samples/zones/share-zone-response.json
+
+
+Delete a Zone Share
+===================
+
+.. rest_method::  DELETE /v2/zones/{zone_id}/shares/{zone_share_id}
+
+Delete a zone share.
+
+**New in version 2.1**
+
+.. rest_status_code:: success status.yaml
+
+   - 204
+
+.. rest_status_code:: error status.yaml
+
+   - 400
+   - 401
+   - 403
+   - 404
+   - 405
+   - 500
+   - 503
+
+Request
+-------
+
+.. rest_parameters:: parameters.yaml
+
+   - x-auth-token: x-auth-token
+   - x-auth-all-projects: x-auth-all-projects
+   - x-auth-sudo-project-id: x-auth-sudo-project-id
+   - zone_id: path_zone_id
+   - zone_share_id: path_zone_share_id
+
+Response Parameters
+-------------------
+
+.. rest_parameters:: parameters.yaml
+
+   - x-openstack-request-id: x-openstack-request-id

api-ref/source/dns-api-v2-zone.inc

@@ -75,6 +75,7 @@ Response Parameters
    - created_at: created_at
    - updated_at: updated_at
    - attributes: zone_attributes
+   - shared: shared
    - links: links
 
 
@@ -152,6 +153,7 @@ Response Parameters
    - created_at: created_at
    - updated_at: updated_at
    - attributes: zone_attributes
+   - shared: shared
    - links: links
    - metadata: metadata
 
@@ -221,6 +223,7 @@ Response Parameters
    - created_at: created_at
    - updated_at: updated_at
    - attributes: zone_attributes
+   - shared: shared
    - links: links
 
 
@@ -352,6 +355,7 @@ Response Parameters
    - created_at: created_at
    - updated_at: updated_at
    - attributes: zone_attributes
+   - shared: shared
    - links: links
 
 
@@ -395,6 +399,7 @@ Request
    - x-auth-all-projects: x-auth-all-projects
    - x-auth-sudo-project-id: x-auth-sudo-project-id
    - x-designate-hard-delete: x-designate-hard-delete
+   - x-designate-delete-shares: x-designate-delete-shares
    - zone_id: path_zone_id
 
 
@@ -421,6 +426,7 @@ Response Parameters
    - created_at: created_at
    - updated_at: updated_at
    - attributes: zone_attributes
+   - shared: shared
    - links: links
 
 

api-ref/source/parameters.yaml

@@ -23,6 +23,14 @@ x-auth-token:
   required: false
   type: string
 
+x-designate-delete-shares:
+  description: |
+    If enabled, this will delete associated shares along with the resource.
+  in: header
+  required: false
+  type: bool
+  min_version: 2.1
+
 x-designate-edit-managed-records:
   description: |
     If enabled this will all users to edit records flagged as managed
@@ -121,6 +129,14 @@ path_zone_import_id:
   required: true
   type: uuid
 
+path_zone_share_id:
+  description: |
+    ID of the zone share.
+  in: path
+  required: true
+  type: uuid
+  min_version: 2.1
+
 path_zone_transfer_accept_id:
   description: |
     ID for this zone transfer accept
@@ -255,6 +271,15 @@ sort_key:
   required: false
   type: string
 
+target_project_id_filter:
+  description: |
+    Filter results to only show resources that have a matching
+    target_project_id
+  in: query
+  required: false
+  type: string
+  min_version: 2.1
+
 tld_name_filter:
   description: |
     Filter results to only show tlds that have a name matching the filter
@@ -691,6 +716,22 @@ service_statuses:
   required: true
   type: array
 
+shared:
+  description: |
+    True if the zone is shared with another project.
+  in: body
+  required: true
+  type: bool
+  min_version: 2.1
+
+shared_zone_id:
+  description: |
+    ID for the zone being shared.
+  in: body
+  required: true
+  type: uuid
+  min_version: 2.1
+
 stats:
   description: |
     Statistics for the service.
@@ -705,6 +746,14 @@ status:
   required: true
   type: enum
 
+target_project_id:
+  description: |
+    The project ID the zone will be shared with.
+  in: body
+  required: true
+  type: string
+  min_version: 2.1
+
 tld_description:
   description: |
     Description for this tld

api-ref/source/samples/zones/list-share-zone-response.json

@@ -0,0 +1,28 @@
+{
+    "shared_zones": [{
+        "id": "4495ffbb-b7d1-43e0-9423-f0a4172e5f9e",
+        "zone_id": "a3365b47-ee93-43ad-9a60-2b2ca96b1898",
+        "project_id": "16ade46c85a1435bb86d9138d37da57e",
+        "target_project_id": "232e37df46af42089710e2ae39111c2f",
+        "created_at": "2022-12-01T23:02:49.000000",
+        "updated_at": null,
+        "links": {
+            "self": "http://127.0.0.1:60053/v2/zones/a3365b47-ee93-43ad-9a60-2b2ca96b1898/shares/4495ffbb-b7d1-43e0-9423-f0a4172e5f9e",
+            "zone": "http://127.0.0.1:60053/v2/zones/a3365b47-ee93-43ad-9a60-2b2ca96b1898"
+        }
+    }, {
+        "id": "1f278d08-2f6a-462a-bb49-21a4f6e6d32b",
+        "zone_id": "a3365b47-ee93-43ad-9a60-2b2ca96b1898",
+        "project_id": "16ade46c85a1435bb86d9138d37da57e",
+        "target_project_id": "86d78e93698e4b06aad4f62e04afb4c1",
+        "created_at": "2022-12-02T01:51:48.000000",
+        "updated_at": null,
+        "links": {
+            "self": "http://127.0.0.1:60053/v2/zones/a3365b47-ee93-43ad-9a60-2b2ca96b1898/shares/1f278d08-2f6a-462a-bb49-21a4f6e6d32b",
+            "zone": "http://127.0.0.1:60053/v2/zones/a3365b47-ee93-43ad-9a60-2b2ca96b1898"
+        }
+    }],
+    "links": {
+        "self": "http://127.0.0.1:60053/v2/zones/a3365b47-ee93-43ad-9a60-2b2ca96b1898/shares"
+    }
+}

api-ref/source/samples/zones/share-zone-request.json

@@ -0,0 +1,3 @@
+{
+    "target_project_id": "232e37df46af42089710e2ae39111c2f"
+}

api-ref/source/samples/zones/share-zone-response.json

@@ -0,0 +1,12 @@
+{
+    "id": "fd40b017-bf97-461c-8d30-d4e922b28edd",
+    "zone_id": "a3365b47-ee93-43ad-9a60-2b2ca96b1898",
+    "project_id": "16ade46c85a1435bb86d9138d37da57e",
+    "target_project_id": "232e37df46af42089710e2ae39111c2f",
+    "created_at": "2022-11-30T22:20:27.000000",
+    "updated_at": null,
+    "links": {
+        "self": "http://127.0.0.1:60053/v2/zones/a3365b47-ee93-43ad-9a60-2b2ca96b1898/shares/fd40b017-bf97-461c-8d30-d4e922b28edd",
+        "zone": "http://127.0.0.1:60053/v2/zones/a3365b47-ee93-43ad-9a60-2b2ca96b1898"
+    }
+}

designate/api/middleware.py

@@ -101,6 +101,14 @@ class ContextMiddleware(base.Middleware):
         if hasattr(request, 'client_addr'):
             ctxt.client_addr = request.client_addr
 
+    @staticmethod
+    def _extract_delete_shares(ctxt, request):
+        ctxt.delete_shares = False
+        if request.headers.get('X-Designate-Delete-Shares'):
+            ctxt.delete_shares = strutils.bool_from_string(
+                request.headers.get('X-Designate-Delete-Shares')
+            )
+
     def make_context(self, request, *args, **kwargs):
         req_id = request.environ.get(request_id.ENV_REQUEST_ID)
         kwargs.setdefault('request_id', req_id)
@@ -114,6 +122,7 @@ class ContextMiddleware(base.Middleware):
             self._extract_hard_delete(ctxt, request)
             self._extract_dns_hide_counts(ctxt, request)
             self._extract_client_addr(ctxt, request)
+            self._extract_delete_shares(ctxt, request)
         finally:
             request.environ['context'] = ctxt
         return ctxt

designate/api/v2/controllers/common.py

@@ -17,11 +17,6 @@ from designate import utils
 
 
 def retrieve_matched_rrsets(context, controller_obj, zone_id, **params):
-    if zone_id:
-        # NOTE: We need to ensure the zone actually exists, otherwise we may
-        #       return deleted recordsets instead of a zone not found
-        controller_obj.central_api.get_zone(context, zone_id)
-
     # Extract the pagination params
     marker, limit, sort_key, sort_dir = utils.get_paging_params(
             context, params, controller_obj.SORT_KEYS)

designate/api/v2/controllers/zones/__init__.py

@@ -20,6 +20,7 @@ import pecan
 from designate.api.v2.controllers import rest
 from designate.api.v2.controllers.zones import nameservers
 from designate.api.v2.controllers.zones import recordsets
+from designate.api.v2.controllers.zones import sharedzones
 from designate.api.v2.controllers.zones import tasks
 from designate import exceptions
 from designate import objects
@@ -40,6 +41,7 @@ class ZonesController(rest.RestController):
     recordsets = recordsets.RecordSetsController()
     tasks = tasks.TasksController()
     nameservers = nameservers.NameServersController()
+    shares = sharedzones.SharedZonesController()
 
     @pecan.expose(template='json:', content_type='application/json')
     @utils.validate_uuid('zone_id')
@@ -102,6 +104,10 @@ class ZonesController(rest.RestController):
         # Create the zone
         zone = self.central_api.create_zone(context, zone)
 
+        # Shared is a virtual database column, so inject False here as a
+        # new zone cannot yet be shared.
+        zone.shared = False
+
         LOG.info("Created %(zone)s", {'zone': zone})
 
         # Prepare the response headers

designate/api/v2/controllers/zones/sharedzones.py

@@ -0,0 +1,110 @@
+# Copyright 2020 Cloudification GmbH. 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 oslo_log import log as logging
+import pecan
+
+from designate.api.v2.controllers import rest
+from designate.common import keystone
+from designate.objects.adapters import DesignateAdapter
+from designate.objects import SharedZone
+from designate import utils
+
+LOG = logging.getLogger(__name__)
+
+
+class SharedZonesController(rest.RestController):
+    SORT_KEYS = ['created_at', 'updated_at', ]
+
+    @pecan.expose(template='json:', content_type='application/json')
+    @utils.validate_uuid('zone_id', 'zone_share_id')
+    def get_one(self, zone_id, zone_share_id):
+        """Get Zone Share"""
+        request = pecan.request
+        context = request.environ['context']
+
+        zone = self.central_api.get_shared_zone(
+                context, zone_id, zone_share_id)
+
+        LOG.info(
+            "Retrieved %(zone)s",
+            {"zone": zone}
+        )
+
+        return DesignateAdapter.render('API_v2', zone, request=request)
+
+    @pecan.expose(template='json:', content_type='application/json')
+    @utils.validate_uuid('zone_id')
+    def get_all(self, zone_id, **params):
+        """List all Shared Zones"""
+        request = pecan.request
+        context = request.environ['context']
+
+        # Extract the pagination params
+        marker, limit, sort_key, sort_dir = utils.get_paging_params(
+            context, params, self.SORT_KEYS)
+
+        # Extract any filter params
+        accepted_filters = ('target_project_id',)
+        criterion = self._apply_filter_params(
+            params, accepted_filters, {})
+
+        criterion['zone_id'] = zone_id
+
+        shared_zones = self.central_api.find_shared_zones(
+            context, criterion, marker, limit, sort_key, sort_dir)
+
+        LOG.info("Retrieved %(shared_zones)s", {'shared_zones': shared_zones})
+
+        return DesignateAdapter.render('API_v2', shared_zones, request=request)
+
+    @pecan.expose(template='json:', content_type='application/json')
+    @utils.validate_uuid('zone_id')
+    def post_all(self, zone_id):
+        """Share Zone"""
+        request = pecan.request
+        response = pecan.response
+        context = request.environ['context']
+
+        payload = request.body_dict
+
+        keystone.verify_project_id(
+            context, payload.get('target_project_id', None)
+        )
+
+        zone_share = DesignateAdapter.parse('API_v2', payload, SharedZone())
+
+        zone_share = self.central_api.share_zone(context, zone_id, zone_share)
+
+        response.status_int = 201
+
+        LOG.info(
+            "Shared %(shared_zone)s",
+            {'shared_zone': zone_share}
+        )
+
+        return DesignateAdapter.render(
+            'API_v2', zone_share, request=request)
+
+    @pecan.expose(template='json:', content_type='application/json')
+    @utils.validate_uuid('zone_id', 'zone_share_id')
+    def delete_one(self, zone_id, zone_share_id):
+        """Unshare Zone"""
+        request = pecan.request
+        response = pecan.response
+        context = request.environ['context']
+
+        zone = self.central_api.unshare_zone(context, zone_id, zone_share_id)
+        response.status_int = 204
+
+        LOG.info("Unshared %(zone)s", {'zone': zone})

designate/api/versions.py

@@ -48,8 +48,11 @@ def factory(global_config, **local_conf):
         # Initial API version for v2 API
         _add_a_version(versions, 'v2', api_url, constants.SUPPORTED,
                        '2022-06-29T00:00:00Z')
-        _add_a_version(versions, 'v2.0', api_url, constants.CURRENT,
+        _add_a_version(versions, 'v2.0', api_url, constants.SUPPORTED,
                        '2022-06-29T00:00:00Z')
+        # 2.1 Shared Zones
+        _add_a_version(versions, 'v2.1', api_url, constants.CURRENT,
+                       '2023-01-25T00:00:00Z')
 
         return flask.jsonify({'versions': versions})
 

designate/central/rpcapi.py

@@ -67,8 +67,9 @@ class CentralAPI(object):
         6.3 - Changed 'update_status' method args
         6.4 - Removed unused record and diagnostic methods
         6.5 - Removed additional unused methods
+        6.6 - Add methods for shared zones
     """
-    RPC_API_VERSION = '6.5'
+    RPC_API_VERSION = '6.6'
 
     # This allows us to mark some methods as not logged.
     # This can be for a few reasons - some methods my not actually call over
@@ -81,7 +82,7 @@ class CentralAPI(object):
 
         target = messaging.Target(topic=self.topic,
                                   version=self.RPC_API_VERSION)
-        self.client = rpc.get_client(target, version_cap='6.5')
+        self.client = rpc.get_client(target, version_cap='6.6')
 
     @classmethod
     def get_instance(cls):
@@ -167,6 +168,27 @@ class CentralAPI(object):
         return self.client.call(context, 'purge_zones',
                                 criterion=criterion, limit=limit)
 
+    # Shared Zone methods
+    def share_zone(self, context, zone_id, shared_zone):
+        return self.client.call(context, 'share_zone', zone_id=zone_id,
+                                shared_zone=shared_zone)
+
+    def unshare_zone(self, context, zone_id, zone_share_id):
+        return self.client.call(context, 'unshare_zone',
+                                zone_id=zone_id, zone_share_id=zone_share_id)
+
+    def find_shared_zones(self, context, criterion=None, marker=None,
+                          limit=None, sort_key=None, sort_dir=None):
+        return self.client.call(
+            context, 'find_shared_zones', criterion=criterion, marker=marker,
+            limit=limit, sort_key=sort_key, sort_dir=sort_dir)
+
+    def get_shared_zone(self, context, zone_id, zone_share_id):
+        return self.client.call(
+            context, 'get_shared_zone', zone_id=zone_id,
+            zone_share_id=zone_share_id
+        )
+
     # TLD Methods
     def create_tld(self, context, tld):
         return self.client.call(context, 'create_tld', tld=tld)

designate/central/service.py

@@ -51,7 +51,7 @@ LOG = logging.getLogger(__name__)
 
 
 class Service(service.RPCService):
-    RPC_API_VERSION = '6.5'
+    RPC_API_VERSION = '6.6'
 
     target = messaging.Target(version=RPC_API_VERSION)
 
@@ -858,21 +858,37 @@ class Service(service.RPCService):
         return zone
 
     @rpc.expected_exceptions()
-    def get_zone(self, context, zone_id):
+    def get_zone(self, context, zone_id, apply_tenant_criteria=True):
         """Get a zone, even if flagged for deletion
         """
-        zone = self.storage.get_zone(context, zone_id)
+        zone = self.storage.get_zone(
+            context, zone_id, apply_tenant_criteria=apply_tenant_criteria)
 
+        # Save a DB round trip if we don't need to check for shared
+        zone_shared = False
+        if (context.project_id != zone.tenant_id) and not context.all_tenants:
+            zone_shared = self.storage.is_zone_shared_with_project(
+                zone_id, context.project_id)
+            if not zone_shared:
+                # Maintain consistency with the previous API and _find_zones()
+                # and _find() when apply_tenant_criteria is True.
+                raise exceptions.ZoneNotFound(
+                    "Could not find %s" % zone.obj_name())
+
+        # TODO(johnsom) This should account for all-projects context
+        # it passes today due to ADMIN
         if policy.enforce_new_defaults():
             target = {
                 'zone_id': zone_id,
                 'zone_name': zone.name,
+                'zone_shared': zone_shared,
                 constants.RBAC_PROJECT_ID: zone.tenant_id
             }
         else:
             target = {
                 'zone_id': zone_id,
                 'zone_name': zone.name,
+                'zone_shared': zone_shared,
                 'tenant_id': zone.tenant_id
             }
 
@@ -1033,6 +1049,14 @@ class Service(service.RPCService):
         else:
             policy.check('delete_zone', context, target)
 
+        # Prevent the deletion of a shared zone if the delete-shares modifier
+        # is not specified.
+        if zone.shared and not context.delete_shares:
+            raise exceptions.ZoneShared(
+                'This zone is shared with other projects, please remove these '
+                'shares before deletion or use the delete-shares modifier to '
+                'override this warning.')
+
         # Prevent deletion of a zone which has child zones
         criterion = {'parent_zone_id': zone_id}
 
@@ -1042,6 +1066,11 @@ class Service(service.RPCService):
             raise exceptions.ZoneHasSubZone('Please delete any subzones '
                                             'before deleting this zone')
 
+        # If the zone is shared and delete_shares was specified, remove all
+        # of the zone shares in preparation for the zone delete.
+        if zone.shared and context.delete_shares:
+            self.storage.delete_zone_shares(zone.id)
+
         if hasattr(context, 'abandon') and context.abandon:
             LOG.info("Abandoning zone '%(zone)s'", {'zone': zone.name})
             zone = self.storage.delete_zone(context, zone.id)
@@ -1165,13 +1194,150 @@ class Service(service.RPCService):
 
         return reports
 
+    # Shared zones
+    @rpc.expected_exceptions()
+    @notification.notify_type('dns.zone.share')
+    @transaction
+    def share_zone(self, context, zone_id, shared_zone):
+        # Ensure that zone exists and get the zone owner
+        zone = self.storage.get_zone(context, zone_id)
+
+        if policy.enforce_new_defaults():
+            target = {constants.RBAC_PROJECT_ID: zone.tenant_id}
+        else:
+            target = {'tenant_id': zone.tenant_id}
+
+        policy.check('share_zone', context, target)
+
+        shared_zone['project_id'] = context.project_id
+        shared_zone['zone_id'] = zone_id
+
+        shared_zone = self.storage.share_zone(context, shared_zone)
+
+        return shared_zone
+
+    @rpc.expected_exceptions()
+    @notification.notify_type('dns.zone.unshare')
+    @transaction
+    def unshare_zone(self, context, zone_id, zone_share_id):
+        # Ensure the share exists and get the share owner
+        shared_zone = self.get_shared_zone(context, zone_id, zone_share_id)
+
+        if policy.enforce_new_defaults():
+            target = {constants.RBAC_PROJECT_ID: shared_zone.project_id}
+        else:
+            target = {'tenant_id': shared_zone.project_id}
+
+        policy.check('unshare_zone', context, target)
+
+        # Prevent unsharing of a zone which has child zones in other tenants
+        criterion = {
+            'parent_zone_id': shared_zone.zone_id,
+            'tenant_id': "%s" % shared_zone.target_project_id,
+        }
+
+        # Look for child zones across all tenants with elevated context
+        if self.storage.count_zones(context.elevated(all_tenants=True),
+                                    criterion) > 0:
+            raise exceptions.SharedZoneHasSubZone(
+                'Please delete all subzones owned by project %s '
+                'before unsharing this zone' % shared_zone.target_project_id
+            )
+
+        # Prevent unsharing of a zone which has recordsets in other tenants
+        criterion = {
+            'zone_id': shared_zone.zone_id,
+            'tenant_id': "%s" % shared_zone.target_project_id,
+        }
+
+        # Look for recordsets across all tenants with elevated context
+        if self.storage.count_recordsets(
+                context.elevated(all_tenants=True), criterion) > 0:
+            raise exceptions.SharedZoneHasRecordSets(
+                'Please delete all recordsets owned by project %s '
+                'before unsharing this zone.' % shared_zone.target_project_id
+            )
+
+        shared_zone = self.storage.unshare_zone(
+            context, zone_id, zone_share_id
+        )
+
+        return shared_zone
+
+    @rpc.expected_exceptions()
+    def find_shared_zones(self, context, criterion=None, marker=None,
+                          limit=None, sort_key=None, sort_dir=None):
+
+        # By default we will let any valid token through as the filter
+        # criteria below will limit the scope of the results.
+        policy.check('find_zone_shares', context)
+
+        if not context.all_tenants and criterion:
+            # Check that they are asking for another projects shares
+            if policy.enforce_new_defaults():
+                target = {constants.RBAC_PROJECT_ID: criterion.get(
+                    'target_project_id', context.project_id)}
+            else:
+                target = {'tenant_id': criterion.get('target_project_id',
+                                                     context.project_id)}
+
+            policy.check('find_project_zone_share', context, target)
+
+        shared_zones = self.storage.find_shared_zones(
+            context, criterion, marker, limit, sort_key, sort_dir
+        )
+
+        return shared_zones
+
+    @rpc.expected_exceptions()
+    def get_shared_zone(self, context, zone_id, zone_share_id):
+        # Ensure that share exists and get the share owner
+        zone_share = self.storage.get_shared_zone(
+            context, zone_id, zone_share_id)
+
+        if policy.enforce_new_defaults():
+            target = {constants.RBAC_PROJECT_ID: zone_share.project_id}
+        else:
+            target = {'tenant_id': zone_share.project_id}
+
+        policy.check('get_zone_share', context, target)
+
+        return zone_share
+
+    def _check_zone_share_permission(self, context, zone):
+        """
+        Check if a request is acceptable for the requesting project ID.
+        If the requestor is not the zone owner and the zone is not shared
+        with them, return a 404 Not Found to match previous API versions.
+        Otherwise, the later RBAC check will raise a 403 Forbidden.
+
+        :param context: The security context for the request.
+        :param zone: The zone the request is against.
+        :return: If the zone is shared with the requesting project ID or not.
+        """
+        zone_shared = False
+        if (context.project_id != zone.tenant_id) and not context.all_tenants:
+            zone_shared = self.storage.is_zone_shared_with_project(
+                zone.id, context.project_id)
+            if not zone_shared:
+                # Maintain consistency with the previous API and _find_zones()
+                # and _find() when apply_tenant_criteria is True.
+                raise exceptions.ZoneNotFound(
+                    "Could not find %s" % zone.obj_name())
+        return zone_shared
+
     # RecordSet Methods
     @rpc.expected_exceptions()
     @notification.notify_type('dns.recordset.create')
     @lock.synchronized_zone()
     def create_recordset(self, context, zone_id, recordset,
                          increment_serial=True):
-        zone = self.storage.get_zone(context, zone_id)
+        zone = self.storage.get_zone(context, zone_id,
+                                     apply_tenant_criteria=False)
+
+        # Note this call must follow the get_zone call to maintain API response
+        # code behavior.
+        zone_shared = self._check_zone_share_permission(context, zone)
 
         # Don't allow updates to zones that are being deleted
         if zone.action == 'DELETE':
@@ -1182,6 +1348,7 @@ class Service(service.RPCService):
                 'zone_id': zone_id,
                 'zone_name': zone.name,
                 'zone_type': zone.type,
+                'zone_shared': zone_shared,
                 'recordset_name': recordset.name,
                 constants.RBAC_PROJECT_ID: zone.tenant_id,
             }
@@ -1190,12 +1357,20 @@ class Service(service.RPCService):
                 'zone_id': zone_id,
                 'zone_name': zone.name,
                 'zone_type': zone.type,
+                'zone_shared': zone_shared,
                 'recordset_name': recordset.name,
                 'tenant_id': zone.tenant_id,
             }
 
         policy.check('create_recordset', context, target)
 
+        # Override the context to be all_tenants here as we have already
+        # passed the RBAC check for this call and context checks in lower
+        # layers will fail for shared zones.
+        # TODO(johnsom) Remove once context checking is removed from the lower
+        #               code layers.
+        context = context.elevated(all_tenants=True)
+
         recordset, zone = self._create_recordset_in_storage(
             context, zone, recordset, increment_serial=increment_serial)
 
@@ -1267,20 +1442,35 @@ class Service(service.RPCService):
 
     @rpc.expected_exceptions()
     def get_recordset(self, context, zone_id, recordset_id):
-        recordset = self.storage.get_recordset(context, recordset_id)
+        # apply_tenant_criteria=False here as we will gate visibility
-
+        # with the RBAC rules below. This allows project that share the zone
+        # to see all of the records of the zone.
         if zone_id:
-            zone = self.storage.get_zone(context, zone_id)
+            recordset = self.storage.find_recordset(
+                context, criterion={'id': recordset_id, 'zone_id': zone_id},
+                apply_tenant_criteria=False)
+            zone = self.storage.get_zone(context, zone_id,
+                                         apply_tenant_criteria=False)
             # Ensure the zone_id matches the record's zone_id
             if zone.id != recordset.zone_id:
                 raise exceptions.RecordSetNotFound()
         else:
-            zone = self.storage.get_zone(context, recordset.zone_id)
+            recordset = self.storage.find_recordset(
+                context, criterion={'id': recordset_id},
+                apply_tenant_criteria=False)
+            zone = self.storage.get_zone(context, recordset.zone_id,
+                                         apply_tenant_criteria=False)
 
+        # Note this call must follow the get_zone call to maintain API response
+        # code behavior.
+        zone_shared = self._check_zone_share_permission(context, zone)
+
+        # TODO(johnsom) This should account for all_projects
         if policy.enforce_new_defaults():
             target = {
                 'zone_id': zone.id,
                 'zone_name': zone.name,
+                'zone_shared': zone_shared,
                 'recordset_id': recordset.id,
                 constants.RBAC_PROJECT_ID: zone.tenant_id,
             }
@@ -1288,6 +1478,7 @@ class Service(service.RPCService):
             target = {
                 'zone_id': zone.id,
                 'zone_name': zone.name,
+                'zone_shared': zone_shared,
                 'recordset_id': recordset.id,
                 'tenant_id': zone.tenant_id,
             }
@@ -1303,6 +1494,19 @@ class Service(service.RPCService):
     @rpc.expected_exceptions()
     def find_recordsets(self, context, criterion=None, marker=None, limit=None,
                         sort_key=None, sort_dir=None, force_index=False):
+        zone = None
+        zone_shared = False
+
+        if criterion and criterion.get('zone_id', None):
+            # NOTE: We need to ensure the zone actually exists, otherwise
+            # we may return deleted recordsets instead of a zone not found
+            zone = self.get_zone(context, criterion['zone_id'],
+                                 apply_tenant_criteria=False)
+            # Note this call must follow the get_zone call to maintain API
+            # response code behavior.
+            zone_shared = self._check_zone_share_permission(context, zone)
+
+        # TODO(johnsom) Fix this to be useful
         if policy.enforce_new_defaults():
             target = {constants.RBAC_PROJECT_ID: context.project_id}
         else:
@@ -1310,14 +1514,22 @@ class Service(service.RPCService):
 
         policy.check('find_recordsets', context, target)
 
-        recordsets = self.storage.find_recordsets(context, criterion, marker,
+        apply_tenant_criteria = True
-                                                  limit, sort_key, sort_dir,
+        # NOTE(imalinovskiy): Show all recordsets for zone owner or if the zone
-                                                  force_index)
+        #                     is shared with this project.
+        if (zone and zone.tenant_id == context.project_id) or zone_shared:
+            apply_tenant_criteria = False
+
+        recordsets = self.storage.find_recordsets(
+            context, criterion, marker, limit, sort_key, sort_dir, force_index,
+            apply_tenant_criteria=apply_tenant_criteria)
 
         return recordsets
 
     @rpc.expected_exceptions()
     def find_recordset(self, context, criterion=None):
+
+        # TODO(johnsom) Fix this to be useful
         if policy.enforce_new_defaults():
             target = {constants.RBAC_PROJECT_ID: context.project_id}
         else:
@@ -1344,8 +1556,6 @@ class Service(service.RPCService):
     @lock.synchronized_zone()
     def update_recordset(self, context, recordset, increment_serial=True):
         zone_id = recordset.obj_get_original_value('zone_id')
-        zone = self.storage.get_zone(context, zone_id)
-
         changes = recordset.obj_get_changes()
 
         # Ensure immutable fields are not changed
@@ -1361,24 +1571,39 @@ class Service(service.RPCService):
             raise exceptions.BadRequest('Changing a recordsets type is not '
                                         'allowed')
 
+        zone = self.storage.get_zone(context, zone_id,
+                                     apply_tenant_criteria=False)
+
+        # Note this call must follow the get_zone call to maintain API response
+        # code behavior.
+        zone_shared = self._check_zone_share_permission(context, zone)
+
         # Don't allow updates to zones that are being deleted
         if zone.action == 'DELETE':
             raise exceptions.BadRequest('Can not update a deleting zone')
 
+        # TODO(johnsom) This should account for all-projects context
+        # it passes today due to ADMIN
         if policy.enforce_new_defaults():
             target = {
-                'zone_id': recordset.obj_get_original_value('zone_id'),
-                'zone_type': zone.type,
                 'recordset_id': recordset.obj_get_original_value('id'),
+                'recordset_project_id': recordset.obj_get_original_value(
+                    'tenant_id'),
+                'zone_id': recordset.obj_get_original_value('zone_id'),
                 'zone_name': zone.name,
+                'zone_shared': zone_shared,
+                'zone_type': zone.type,
                 constants.RBAC_PROJECT_ID: zone.tenant_id
             }
         else:
             target = {
-                'zone_id': recordset.obj_get_original_value('zone_id'),
-                'zone_type': zone.type,
                 'recordset_id': recordset.obj_get_original_value('id'),
+                'recordset_project_id': recordset.obj_get_original_value(
+                    'tenant_id'),
+                'zone_id': recordset.obj_get_original_value('zone_id'),
                 'zone_name': zone.name,
+                'zone_shared': zone_shared,
+                'zone_type': zone.type,
                 'tenant_id': zone.tenant_id
             }
 
@@ -1387,6 +1612,13 @@ class Service(service.RPCService):
         if recordset.managed and not context.edit_managed_records:
             raise exceptions.BadRequest('Managed records may not be updated')
 
+        # Override the context to be all_tenants here as we have already
+        # passed the RBAC check for this call and context checks in lower
+        # layers will fail for shared zones.
+        # TODO(johnsom) Remove once context checking is removed from the lower
+        #               code layers.
+        context = context.elevated(all_tenants=True)
+
         recordset, zone = self._update_recordset_in_storage(
             context, zone, recordset, increment_serial=increment_serial)
 
@@ -1427,23 +1659,29 @@ class Service(service.RPCService):
     @lock.synchronized_zone()
     def delete_recordset(self, context, zone_id, recordset_id,
                          increment_serial=True):
-        zone = self.storage.get_zone(context, zone_id)
+        # apply_tenant_criteria=False here as we will gate this delete
-        recordset = self.storage.get_recordset(context, recordset_id)
+        # with the RBAC rules below. This allows the zone owner to delete
-
+        # all of the recordsets of the zone.
-        # Ensure the zone_id matches the recordset's zone_id
+        recordset = self.storage.find_recordset(
-        if zone.id != recordset.zone_id:
+            context,
-            raise exceptions.RecordSetNotFound()
+            {"id": recordset_id, "zone_id": zone_id},
+            apply_tenant_criteria=False
+        )
+        zone = self.storage.get_zone(context, zone_id,
+                                     apply_tenant_criteria=False)
 
         # Don't allow updates to zones that are being deleted
         if zone.action == 'DELETE':
             raise exceptions.BadRequest('Can not update a deleting zone')
 
+        # TODO(johnsom) should handle all_projects
         if policy.enforce_new_defaults():
             target = {
                 'zone_id': zone_id,
                 'zone_name': zone.name,
                 'zone_type': zone.type,
                 'recordset_id': recordset.id,
+                'recordset_project_id': recordset.tenant_id,
                 constants.RBAC_PROJECT_ID: zone.tenant_id
             }
         else:
@@ -1452,6 +1690,7 @@ class Service(service.RPCService):
                 'zone_name': zone.name,
                 'zone_type': zone.type,
                 'recordset_id': recordset.id,
+                'recordset_project_id': recordset.tenant_id,
                 'tenant_id': zone.tenant_id
             }
 
@@ -1460,6 +1699,12 @@ class Service(service.RPCService):
         if recordset.managed and not context.edit_managed_records:
             raise exceptions.BadRequest('Managed records may not be deleted')
 
+        # Override the context to be all_tenants here as we have already
+        # passed the RBAC check for this call.
+        # TODO(johnsom) Remove once context checking is removed from the lower
+        #               code layers.
+        context = context.elevated(all_tenants=True)
+
         recordset, zone = self._delete_recordset_in_storage(
             context, zone, recordset, increment_serial=increment_serial)
 
@@ -1797,8 +2042,8 @@ class Service(service.RPCService):
 
         if not recordset:
             try:
-                recordset = self.storage.get_recordset(
+                recordset = self.storage.find_recordset(
-                    elevated_context, record.recordset_id
+                    elevated_context, criterion={'id': record.recordset_id}
                 )
             except exceptions.RecordSetNotFound:
                 LOG.debug('No recordset found for %s', fip['id'])

designate/common/policies/__init__.py

@@ -25,6 +25,7 @@ from designate.common.policies import quota
 from designate.common.policies import record
 from designate.common.policies import recordset
 from designate.common.policies import service_status
+from designate.common.policies import shared_zones
 from designate.common.policies import tenant
 from designate.common.policies import tld
 from designate.common.policies import tsigkey
@@ -45,6 +46,7 @@ def list_rules():
         record.list_rules(),
         recordset.list_rules(),
         service_status.list_rules(),
+        shared_zones.list_rules(),
         tenant.list_rules(),
         tld.list_rules(),
         tsigkey.list_rules(),

designate/common/policies/base.py

@@ -66,6 +66,10 @@ SYSTEM_OR_PROJECT_READER_OR_ALL_TENANTS_READER = (
     ALL_TENANTS_READER + ')'
 )
 
+SYSTEM_OR_PROJECT_READER_OR_SHARED = (
+        SYSTEM_OR_PROJECT_READER + ' or ("True":%(zone_shared)s)'
+)
+
 RULE_ZONE_TRANSFER = (
     '(' + SYSTEM_ADMIN_OR_PROJECT_MEMBER + ') or '
     'project_id:%(target_project_id)s or '
@@ -79,6 +83,9 @@ RULE_ADMIN_OR_OWNER = 'rule:admin_or_owner'
 LEGACY_RULE_ZONE_TRANSFER = "rule:admin_or_owner OR " \
                             "project_id:%(target_tenant_id)s " \
                             "OR None:%(target_tenant_id)s"
+RULE_ADMIN_OR_OWNER_OR_SHARED = (
+        RULE_ADMIN_OR_OWNER + ' or ("True":%(zone_shared)s)'
+)
 
 deprecated_default = policy.DeprecatedRule(
     name="default",

designate/common/policies/recordset.py

@@ -28,9 +28,20 @@ RULE_ZONE_PRIMARY_OR_ADMIN = (
     "('PRIMARY':%(zone_type)s and rule:admin_or_owner) "
     "OR ('SECONDARY':%(zone_type)s AND is_admin:True)")
 
+RULE_ZONE_PRIMARY_OR_ADMIN_OR_SHARED = (
+    "('PRIMARY':%(zone_type)s AND (rule:admin_or_owner OR "
+    "'True':%(zone_shared)s)) "
+    "OR ('SECONDARY':%(zone_type)s AND is_admin:True)")
+
+RULE_ADMIN_OR_OWNER_PRIMARY = (
+    "rule:admin or (\'PRIMARY\':%(zone_type)s and "
+    "(rule:owner or project_id:%(recordset_project_id)s))"
+)
+
+
 deprecated_create_recordset = policy.DeprecatedRule(
     name="create_recordset",
-    check_str=RULE_ZONE_PRIMARY_OR_ADMIN,
+    check_str=RULE_ZONE_PRIMARY_OR_ADMIN_OR_SHARED,
     deprecated_reason=DEPRECATED_REASON,
     deprecated_since=versionutils.deprecated.WALLABY
 )
@@ -42,7 +53,7 @@ deprecated_get_recordsets = policy.DeprecatedRule(
 )
 deprecated_get_recordset = policy.DeprecatedRule(
     name="get_recordset",
-    check_str=base.RULE_ADMIN_OR_OWNER,
+    check_str=base.RULE_ADMIN_OR_OWNER_OR_SHARED,
     deprecated_reason=DEPRECATED_REASON,
     deprecated_since=versionutils.deprecated.WALLABY
 )
@@ -60,13 +71,13 @@ deprecated_find_recordsets = policy.DeprecatedRule(
 )
 deprecated_update_recordset = policy.DeprecatedRule(
     name="update_recordset",
-    check_str=RULE_ZONE_PRIMARY_OR_ADMIN,
+    check_str=RULE_ADMIN_OR_OWNER_PRIMARY,
     deprecated_reason=DEPRECATED_REASON,
     deprecated_since=versionutils.deprecated.WALLABY
 )
 deprecated_delete_recordset = policy.DeprecatedRule(
     name="delete_recordset",
-    check_str=RULE_ZONE_PRIMARY_OR_ADMIN,
+    check_str=RULE_ADMIN_OR_OWNER_PRIMARY,
     deprecated_reason=DEPRECATED_REASON,
     deprecated_since=versionutils.deprecated.WALLABY
 )
@@ -86,11 +97,27 @@ SYSTEM_ADMIN_AND_PRIMARY_ZONE = (
 SYSTEM_ADMIN_AND_SECONDARY_ZONE = (
     '(' + base.SYSTEM_ADMIN + ') and (\'SECONDARY\':%(zone_type)s)'
 )
+SHARED_AND_PRIMARY_ZONE = (
+    '("True":%(zone_shared)s) and (\'PRIMARY\':%(zone_type)s)'
+)
+RECORDSET_MEMBER_AND_PRIMARY_ZONE = (
+    'role:member and (project_id:%(recordset_project_id)s) and '
+    '(\'PRIMARY\':%(zone_type)s)'
+)
+
 
 SYSTEM_ADMIN_OR_PROJECT_MEMBER_ZONE_TYPE = ' or '.join(
     [PROJECT_MEMBER_AND_PRIMARY_ZONE,
      SYSTEM_ADMIN_AND_PRIMARY_ZONE,
-     SYSTEM_ADMIN_AND_SECONDARY_ZONE]
+     SYSTEM_ADMIN_AND_SECONDARY_ZONE,
+     SHARED_AND_PRIMARY_ZONE]
+)
+
+SYSTEM_ADMIN_OR_PROJECT_MEMBER_RECORD_OWNER_ZONE_TYPE = ' or '.join(
+    [PROJECT_MEMBER_AND_PRIMARY_ZONE,
+     SYSTEM_ADMIN_AND_PRIMARY_ZONE,
+     SYSTEM_ADMIN_AND_SECONDARY_ZONE,
+     RECORDSET_MEMBER_AND_PRIMARY_ZONE]
 )
 
 
@@ -116,7 +143,7 @@ rules = [
     ),
     policy.DocumentedRuleDefault(
         name="get_recordset",
-        check_str=base.SYSTEM_OR_PROJECT_READER,
+        check_str=base.SYSTEM_OR_PROJECT_READER_OR_SHARED,
         scope_types=['system', 'project'],
         description="Get recordset",
         operations=[
@@ -149,7 +176,7 @@ rules = [
     ),
     policy.DocumentedRuleDefault(
         name="update_recordset",
-        check_str=SYSTEM_ADMIN_OR_PROJECT_MEMBER_ZONE_TYPE,
+        check_str=SYSTEM_ADMIN_OR_PROJECT_MEMBER_RECORD_OWNER_ZONE_TYPE,
         scope_types=['system', 'project'],
         description="Update recordset",
         operations=[
@@ -162,7 +189,7 @@ rules = [
     ),
     policy.DocumentedRuleDefault(
         name="delete_recordset",
-        check_str=SYSTEM_ADMIN_OR_PROJECT_MEMBER_ZONE_TYPE,
+        check_str=SYSTEM_ADMIN_OR_PROJECT_MEMBER_RECORD_OWNER_ZONE_TYPE,
         scope_types=['system', 'project'],
         description="Delete RecordSet",
         operations=[
@@ -178,7 +205,7 @@ rules = [
         check_str=base.SYSTEM_OR_PROJECT_READER,
         scope_types=['system', 'project'],
         description="Count recordsets",
-        deprecated_rule=deprecated_count_recordset
+        deprecated_rule=deprecated_count_recordset,
     )
 ]
 

designate/common/policies/shared_zones.py

@@ -0,0 +1,116 @@
+# 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 oslo_log import versionutils
+from oslo_policy import policy
+
+from designate.common.policies import base
+
+
+DEPRECATED_REASON = """
+The shared zones API now supports system scope and default roles.
+"""
+
+deprecated_get_shared_zone = policy.DeprecatedRule(
+    name="get_zone_share",
+    check_str=base.RULE_ADMIN_OR_OWNER,
+    deprecated_reason=DEPRECATED_REASON,
+    deprecated_since=versionutils.deprecated.WALLABY
+)
+
+deprecated_share_zone = policy.DeprecatedRule(
+    name="share_zone",
+    check_str=base.RULE_ADMIN_OR_OWNER,
+    deprecated_reason=DEPRECATED_REASON,
+    deprecated_since=versionutils.deprecated.WALLABY
+)
+
+deprecated_find_project_zone_share = policy.DeprecatedRule(
+    name="find_project_zone_share",
+    check_str=base.RULE_ADMIN_OR_OWNER,
+    deprecated_reason=DEPRECATED_REASON,
+    deprecated_since=versionutils.deprecated.WALLABY
+)
+
+deprecated_unshare_zone = policy.DeprecatedRule(
+    name="unshare_zone",
+    check_str=base.RULE_ADMIN_OR_OWNER,
+    deprecated_reason=DEPRECATED_REASON,
+    deprecated_since=versionutils.deprecated.WALLABY
+)
+
+rules = [
+    policy.DocumentedRuleDefault(
+        name="get_zone_share",
+        check_str=base.SYSTEM_ADMIN_OR_PROJECT_MEMBER,
+        scope_types=['system', 'project'],
+        description="Get a Zone Share",
+        operations=[
+            {
+                'path': '/v2/zones/{zone_id}/shares/{zone_share_id}',
+                'method': 'GET'
+            }
+        ],
+        deprecated_rule=deprecated_get_shared_zone
+    ),
+    policy.DocumentedRuleDefault(
+        name="share_zone",
+        check_str=base.SYSTEM_ADMIN_OR_PROJECT_MEMBER,
+        scope_types=['system', 'project'],
+        description="Share a Zone",
+        operations=[
+            {
+                'path': '/v2/zones/{zone_id}/shares',
+                'method': 'POST'
+            }
+        ],
+        deprecated_rule=deprecated_share_zone
+    ),
+    policy.DocumentedRuleDefault(
+        name="find_zone_shares",
+        # Using rule ANY here because the search criteria will narrow the
+        # results appropriate for the API call.
+        check_str=base.RULE_ANY,
+        description="List Shared Zones",
+        operations=[
+            {
+                'path': '/v2/zones/{zone_id}/shares',
+                'method': 'GET'
+            }
+        ]
+    ),
+    policy.RuleDefault(
+        name="find_project_zone_share",
+        check_str=base.SYSTEM_ADMIN_OR_PROJECT_MEMBER,
+        scope_types=['system', 'project'],
+        description="Check the can query for a specific projects shares.",
+        deprecated_rule=deprecated_find_project_zone_share
+    ),
+    policy.DocumentedRuleDefault(
+        name="unshare_zone",
+        check_str=base.SYSTEM_ADMIN_OR_PROJECT_MEMBER,
+        scope_types=['system', 'project'],
+        description="Unshare Zone",
+        operations=[
+            {
+                'path': '/v2/zones/{zone_id}/shares/{shared_zone_id}',
+                'method': 'DELETE'
+            }
+        ],
+        deprecated_rule=deprecated_unshare_zone
+    )
+]
+
+
+def list_rules():
+    return rules

designate/common/policies/zone.py

@@ -36,7 +36,7 @@ deprecated_get_zones = policy.DeprecatedRule(
 )
 deprecated_get_zone = policy.DeprecatedRule(
     name="get_zone",
-    check_str=base.RULE_ADMIN_OR_OWNER,
+    check_str=base.RULE_ADMIN_OR_OWNER_OR_SHARED,
     deprecated_reason=DEPRECATED_REASON,
     deprecated_since=versionutils.deprecated.WALLABY
 )
@@ -124,7 +124,7 @@ rules = [
     ),
     policy.DocumentedRuleDefault(
         name="get_zone",
-        check_str=base.SYSTEM_OR_PROJECT_READER,
+        check_str=base.SYSTEM_OR_PROJECT_READER_OR_SHARED,
         scope_types=['system', 'project'],
         description="Get Zone",
         operations=[

designate/context.py

@@ -34,17 +34,18 @@ class DesignateContext(context.RequestContext):
     _edit_managed_records = False
     _hard_delete = False
     _client_addr = None
+    _delete_shares = False
     FROM_DICT_EXTRA_KEYS = [
         'original_project_id', 'service_catalog', 'all_tenants', 'abandon',
         'edit_managed_records', 'tsigkey_id', 'hide_counts', 'client_addr',
-        'hard_delete'
+        'hard_delete', 'delete_shares'
     ]
 
     def __init__(self, service_catalog=None, all_tenants=False, abandon=None,
                  tsigkey_id=None, original_project_id=None,
                  edit_managed_records=False, hide_counts=False,
                  client_addr=None, user_auth_plugin=None,
-                 hard_delete=False, **kwargs):
+                 hard_delete=False, delete_shares=False, **kwargs):
         super(DesignateContext, self).__init__(**kwargs)
 
         self.user_auth_plugin = user_auth_plugin
@@ -59,6 +60,7 @@ class DesignateContext(context.RequestContext):
         self.hard_delete = hard_delete
         self.hide_counts = hide_counts
         self.client_addr = client_addr
+        self.delete_shares = delete_shares
 
     def deepcopy(self):
         d = self.to_dict()
@@ -103,6 +105,7 @@ class DesignateContext(context.RequestContext):
             'tsigkey_id': self.tsigkey_id,
             'hide_counts': self.hide_counts,
             'client_addr': self.client_addr,
+            'delete_shares': self.delete_shares,
         })
 
         return copy.deepcopy(d)
@@ -208,6 +211,14 @@ class DesignateContext(context.RequestContext):
     def client_addr(self, value):
         self._client_addr = value
 
+    @property
+    def delete_shares(self):
+        return self._delete_shares
+
+    @delete_shares.setter
+    def delete_shares(self, value):
+        self._delete_shares = value
+
     def get_auth_plugin(self):
         if self.user_auth_plugin:
             return self.user_auth_plugin

designate/exceptions.py

@@ -238,6 +238,18 @@ class ZoneHasSubZone(DesignateException):
     error_type = 'zone_has_sub_zone'
 
 
+class SharedZoneHasSubZone(DesignateException):
+    error_code = 400
+    error_type = 'shared_zone_has_sub_zone'
+    expected = True
+
+
+class SharedZoneHasRecordSets(DesignateException):
+    error_code = 400
+    error_type = 'shared_zone_has_recordsets'
+    expected = True
+
+
 class Forbidden(DesignateException):
     error_code = 403
     error_type = 'forbidden'
@@ -364,6 +376,10 @@ class DuplicateZoneMaster(Duplicate):
     error_type = 'duplicate_zone_attribute'
 
 
+class DuplicateSharedZone(Duplicate):
+    error_type = 'duplicate_shared_zone'
+
+
 class NotFound(DesignateException):
     expected = True
     error_code = 404
@@ -470,6 +486,10 @@ class ZoneExportNotFound(NotFound):
     error_type = 'zone_export_not_found'
 
 
+class SharedZoneNotFound(NotFound):
+    error_type = 'shared_zone_not_found'
+
+
 class LastServerDeleteNotAllowed(BadRequest):
     error_type = 'last_server_delete_not_allowed'
 
@@ -486,3 +506,8 @@ class MissingProjectID(BadRequest):
     # designate/api/middleware.py#L132
     error_code = 401
     error_type = 'missing_project_id'
+
+
+class ZoneShared(DesignateException):
+    error_code = 400
+    error_type = 'zone_is_shared'

designate/objects/__init__.py

@@ -34,6 +34,7 @@ from designate.objects.record import Record, RecordList  # noqa
 from designate.objects.recordset import RecordSet, RecordSetList  # noqa
 from designate.objects.server import Server, ServerList  # noqa
 from designate.objects.service_status import ServiceStatus, ServiceStatusList  # noqa
+from designate.objects.shared_zone import SharedZone, SharedZoneList  # noqa
 from designate.objects.tenant import Tenant, TenantList  # noqa
 from designate.objects.tld import Tld, TldList  # noqa
 from designate.objects.tsigkey import TsigKey, TsigKeyList  # noqa

designate/objects/adapters/__init__.py

@@ -33,7 +33,7 @@ from designate.objects.adapters.api_v2.zone_transfer_request import ZoneTransfer
 from designate.objects.adapters.api_v2.validation_error import ValidationErrorAPIv2Adapter, ValidationErrorListAPIv2Adapter  # noqa
 from designate.objects.adapters.api_v2.zone_import import ZoneImportAPIv2Adapter, ZoneImportListAPIv2Adapter  # noqa
 from designate.objects.adapters.api_v2.zone_export import ZoneExportAPIv2Adapter, ZoneExportListAPIv2Adapter  # noqa
-
+from designate.objects.adapters.api_v2.shared_zone import SharedZoneAPIv2Adapter, SharedZoneListAPIv2Adapter  # noqa
 # YAML
 
 from designate.objects.adapters.yaml.pool import PoolYAMLAdapter, PoolListYAMLAdapter  # noqa

designate/objects/adapters/api_v2/shared_zone.py

@@ -0,0 +1,82 @@
+# Copyright 2020 Cloudification GmbH. 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 urllib import parse
+
+from designate import objects
+from designate.objects.adapters.api_v2 import base
+
+
+class SharedZoneAPIv2Adapter(base.APIv2Adapter):
+
+    ADAPTER_OBJECT = objects.SharedZone
+
+    MODIFICATIONS = {
+        'fields': {
+            "id": {},
+            "zone_id": {},
+            "project_id": {},
+            "target_project_id": {'immutable': True},
+            "created_at": {},
+            "updated_at": {},
+        },
+        'options': {
+            'links': True,
+            'resource_name': 'shared_zone',
+            'collection_name': 'shared_zones',
+        }
+    }
+
+    @classmethod
+    def render_object(cls, object, *args, **kwargs):
+        obj = super(SharedZoneAPIv2Adapter, cls).render_object(
+            object, *args, **kwargs)
+
+        if obj['zone_id'] is not None:
+            obj['links']['self'] = (
+                '%s/v2/zones/%s/shares/%s' % (
+                    cls._get_base_url(kwargs['request']), obj['zone_id'],
+                    obj['id']))
+            obj['links']['zone'] = (
+                '%s/v2/zones/%s' % (cls._get_base_url(kwargs['request']),
+                                    obj['zone_id']))
+        return obj
+
+
+class SharedZoneListAPIv2Adapter(base.APIv2Adapter):
+
+    ADAPTER_OBJECT = objects.SharedZoneList
+
+    MODIFICATIONS = {
+        'options': {
+            'links': True,
+            'resource_name': 'shared_zone',
+            'collection_name': 'shared_zones',
+        }
+    }
+
+    @classmethod
+    def _get_collection_href(cls, request, extra_params=None):
+        params = request.GET
+
+        if extra_params is not None:
+            params.update(extra_params)
+
+        base_uri = cls._get_base_url(request)
+
+        href = "%s%s?%s" % (
+            base_uri,
+            request.path,
+            parse.urlencode(params))
+
+        return href.rstrip('?')

designate/objects/adapters/api_v2/zone.py

@@ -37,6 +37,7 @@ class ZoneAPIv2Adapter(base.APIv2Adapter):
                 'read_only': False
             },
             "serial": {},
+            "shared": {},
             "status": {},
             "action": {},
             "version": {},

designate/objects/shared_zone.py

@@ -0,0 +1,38 @@
+# Copyright 2020 Cloudification GmbH. 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 designate.objects import base
+from designate.objects import fields
+
+
+@base.DesignateRegistry.register
+class SharedZone(base.DictObjectMixin, base.PersistentObjectMixin,
+                 base.DesignateObject):
+    fields = {
+        'zone_id': fields.UUIDFields(nullable=False),
+        'project_id': fields.StringFields(maxLength=36, nullable=False),
+        'target_project_id': fields.StringFields(maxLength=36, nullable=False),
+    }
+
+    STRING_KEYS = [
+        'id', 'zone_id', 'project_id', 'target_project_id'
+    ]
+
+
+@base.DesignateRegistry.register
+class SharedZoneList(base.AttributeListObjectMixin, base.DesignateObject):
+    LIST_ITEM_TYPE = SharedZone
+
+    fields = {
+        'objects': fields.ListOfObjectsField('SharedZone'),
+    }

designate/objects/zone.py

@@ -59,6 +59,7 @@ class Zone(base.DesignateObject, base.DictObjectMixin,
         'recordsets': fields.ObjectField('RecordSetList', nullable=True),
         'attributes': fields.ObjectField('ZoneAttributeList', nullable=True),
         'masters': fields.ObjectField('ZoneMasterList', nullable=True),
+        'shared': fields.BooleanField(default=False, nullable=True),
         'type': fields.EnumField(nullable=True,
                                  valid_values=['SECONDARY', 'PRIMARY'],
                                  read_only=False

designate/storage/base.py

@@ -233,12 +233,13 @@ class Storage(DriverPlugin, metaclass=abc.ABCMeta):
         """
 
     @abc.abstractmethod
-    def get_zone(self, context, zone_id):
+    def get_zone(self, context, zone_id, apply_tenant_criteria=True):
         """
         Get a Zone via its ID.
 
         :param context: RPC Context.
         :param zone_id: ID of the Zone.
+        :param apply_tenant_criteria: Whether to filter results by project_id.
         """
 
     @abc.abstractmethod
@@ -303,29 +304,28 @@ class Storage(DriverPlugin, metaclass=abc.ABCMeta):
         """
 
     @abc.abstractmethod
-    def create_recordset(self, context, zone_id, recordset):
+    def share_zone(self, context, shared_zone):
         """
-        Create a recordset on a given Zone ID
+        Share zone
 
         :param context: RPC Context.
-        :param zone_id: Zone ID to create the recordset in.
+        :param shared_zone: Shared Zone dict
-        :param recordset: RecordSet object with the values to be created.
         """
 
     @abc.abstractmethod
-    def get_recordset(self, context, recordset_id):
+    def unshare_zone(self, context, zone_id, shared_zone_id):
         """
-        Get a recordset via ID
+        Unshare zone
 
         :param context: RPC Context.
-        :param recordset_id: RecordSet ID to get
+        :param shared_zone_id: Shared Zone Id
         """
 
     @abc.abstractmethod
-    def find_recordsets(self, context, criterion=None, marker=None, limit=None,
+    def find_shared_zones(self, context, criterion=None, marker=None,
-                        sort_key=None, sort_dir=None, force_index=False):
+                          limit=None, sort_key=None, sort_dir=None):
         """
-        Find RecordSets.
+        Find shared zones
 
         :param context: RPC Context.
         :param criterion: Criteria to filter by.
@@ -337,6 +337,61 @@ class Storage(DriverPlugin, metaclass=abc.ABCMeta):
         :param sort_dir: Direction to sort after using sort_key.
         """
 
+    @abc.abstractmethod
+    def get_shared_zone(self, context, zone_id, shared_zone_id):
+        """
+        Get a shared zone via ID
+
+        :param context: RPC Context.
+        :param shared_zone_id: Shared Zone Id
+        """
+
+    @abc.abstractmethod
+    def is_zone_shared_with_project(self, zone_id, project_id):
+        """
+        Checks if a zone is shared with a project.
+
+        :param zone_id: The zone ID to check.
+        :param project_id: The project ID to check.
+        :returns: Boolean True/False if the zone is shared with the project.
+        """
+
+    @abc.abstractmethod
+    def delete_zone_shares(self, zone_id):
+        """
+        Delete all of the zone shares for a specific zone.
+
+        :param zone_id: The zone ID to check.
+        """
+
+    @abc.abstractmethod
+    def create_recordset(self, context, zone_id, recordset):
+        """
+        Create a recordset on a given Zone ID
+
+        :param context: RPC Context.
+        :param zone_id: Zone ID to create the recordset in.
+        :param recordset: RecordSet object with the values to be created.
+        """
+
+    @abc.abstractmethod
+    def find_recordsets(self, context, criterion=None, marker=None, limit=None,
+                        sort_key=None, sort_dir=None, force_index=False,
+                        apply_tenant_criteria=True):
+        """
+        Find RecordSets.
+
+        :param context: RPC Context.
+        :param criterion: Criteria to filter by.
+        :param marker: Resource ID from which after the requested page will
+                       start after
+        :param limit: Integer limit of objects of the page size after the
+                      marker
+        :param sort_key: Key from which to sort after.
+        :param sort_dir: Direction to sort after using sort_key.
+        :param apply_tenant_criteria: Whether to filter results by project_id.
+        """
+
     @abc.abstractmethod
     def find_recordsets_axfr(self, context, criterion=None):
         """
@@ -347,12 +402,13 @@ class Storage(DriverPlugin, metaclass=abc.ABCMeta):
         """
 
     @abc.abstractmethod
-    def find_recordset(self, context, criterion):
+    def find_recordset(self, context, criterion, apply_tenant_criteria=True):
         """
         Find a single RecordSet.
 
         :param context: RPC Context.
         :param criterion: Criteria to filter by.
+        :param apply_tenant_criteria: Whether to filter results by project_id.
         """
 
     @abc.abstractmethod

designate/storage/impl_sqlalchemy/__init__.py

@@ -15,8 +15,8 @@
 # under the License.
 from oslo_log import log as logging
 from oslo_utils.secretutils import md5
-from sqlalchemy import select, distinct, func
+from sqlalchemy import case, select, distinct, func
-from sqlalchemy.sql.expression import or_
+from sqlalchemy.sql.expression import or_, literal_column
 
 from designate import exceptions
 from designate import objects
@@ -213,14 +213,23 @@ class SQLAlchemyStorage(sqlalchemy_base.SQLAlchemy, storage_base.Storage):
     # Zone Methods
     ##
     def _find_zones(self, context, criterion, one=False, marker=None,
-                    limit=None, sort_key=None, sort_dir=None):
+                    limit=None, sort_key=None, sort_dir=None,
+                    apply_tenant_criteria=True):
         # Check to see if the criterion can use the reverse_name column
         criterion = self._rname_check(criterion)
 
+        # Create a virtual column showing if the zone is shared or not.
+        shared_case = case((tables.shared_zones.c.target_project_id.is_(None),
+                            literal_column('False')),
+                           else_=literal_column('True')).label('shared')
+        query = select(
+            [tables.zones, shared_case]).outerjoin(tables.shared_zones)
+
         zones = self._find(
             context, tables.zones, objects.Zone, objects.ZoneList,
             exceptions.ZoneNotFound, criterion, one, marker, limit,
-            sort_key, sort_dir)
+            sort_key, sort_dir, query=query,
+            apply_tenant_criteria=apply_tenant_criteria)
 
         def _load_relations(zone):
             if zone.type == 'SECONDARY':
@@ -274,8 +283,9 @@ class SQLAlchemyStorage(sqlalchemy_base.SQLAlchemy, storage_base.Storage):
 
         return zone
 
-    def get_zone(self, context, zone_id):
+    def get_zone(self, context, zone_id, apply_tenant_criteria=True):
-        zone = self._find_zones(context, {'id': zone_id}, one=True)
+        zone = self._find_zones(context, {'id': zone_id}, one=True,
+                                apply_tenant_criteria=apply_tenant_criteria)
         return zone
 
     def find_zones(self, context, criterion=None, marker=None, limit=None,
@@ -504,6 +514,76 @@ class SQLAlchemyStorage(sqlalchemy_base.SQLAlchemy, storage_base.Storage):
 
         return result[0]
 
+    # Shared zones methods
+    def _find_shared_zones(self, context, criterion, one=False, marker=None,
+                           limit=None, sort_key=None, sort_dir=None):
+
+        table = tables.shared_zones
+
+        query = select(table)
+
+        if not context.all_tenants:
+            query = query.where(or_(
+                table.c.project_id == context.project_id,
+                table.c.target_project_id == context.project_id))
+
+        return self._find(
+            context, tables.shared_zones, objects.SharedZone,
+            objects.SharedZoneList, exceptions.SharedZoneNotFound, criterion,
+            one, marker, limit, sort_key, sort_dir, query=query,
+            apply_tenant_criteria=False)
+
+    def _find_zone_share(self, context, zone):
+        criterion = {
+            "target_project_id": context.project_id,
+            "zone_id": zone.id
+        }
+
+        try:
+            return self._find(
+                context, tables.shared_zones, objects.SharedZone,
+                objects.SharedZoneList, exceptions.SharedZoneNotFound,
+                criterion,
+                one=True
+            )
+        except exceptions.SharedZoneNotFound:
+            return None
+
+    def share_zone(self, context, shared_zone):
+        return self._create(tables.shared_zones, shared_zone,
+                            exceptions.DuplicateSharedZone)
+
+    def unshare_zone(self, context, zone_id, shared_zone_id):
+        shared_zone = self._find_shared_zones(
+            context, {'id': shared_zone_id, 'zone_id': zone_id}, one=True
+        )
+        return self._delete(context, tables.shared_zones, shared_zone,
+                            exceptions.SharedZoneNotFound)
+
+    def find_shared_zones(self, context, criterion=None, marker=None,
+                          limit=None, sort_key=None, sort_dir=None):
+        return self._find_shared_zones(
+            context, criterion, marker=marker,
+            limit=limit, sort_key=sort_key, sort_dir=sort_dir
+        )
+
+    def get_shared_zone(self, context, zone_id, shared_zone_id):
+        return self._find_shared_zones(
+            context, {'id': shared_zone_id, 'zone_id': zone_id}, one=True
+        )
+
+    def is_zone_shared_with_project(self, zone_id, project_id):
+        query = select(literal_column('true'))
+        query = query.where(tables.shared_zones.c.zone_id == zone_id)
+        query = query.where(
+            tables.shared_zones.c.target_project_id == project_id)
+        return self.session.scalar(query) is not None
+
+    def delete_zone_shares(self, zone_id):
+        query = tables.shared_zones.delete().where(
+            tables.shared_zones.c.zone_id == zone_id)
+        self.session.execute(query)
+
     # Zone attribute methods
     def _find_zone_attributes(self, context, criterion, one=False,
                               marker=None, limit=None, sort_key=None,
@@ -576,7 +656,7 @@ class SQLAlchemyStorage(sqlalchemy_base.SQLAlchemy, storage_base.Storage):
     # RecordSet Methods
     def _find_recordsets(self, context, criterion, one=False, marker=None,
                          limit=None, sort_key=None, sort_dir=None,
-                         force_index=False):
+                         force_index=False, apply_tenant_criteria=True):
 
         # Check to see if the criterion can use the reverse_name column
         criterion = self._rname_check(criterion)
@@ -598,10 +678,14 @@ class SQLAlchemyStorage(sqlalchemy_base.SQLAlchemy, storage_base.Storage):
             recordsets = self._find(
                 context, tables.recordsets, objects.RecordSet,
                 objects.RecordSetList, exceptions.RecordSetNotFound, criterion,
-                one, marker, limit, sort_key, sort_dir, query)
+                one, marker, limit, sort_key, sort_dir, query,
+                apply_tenant_criteria=apply_tenant_criteria,
+            )
 
             recordsets.records = self._find_records(
-                context, {'recordset_id': recordsets.id})
+                context, {'recordset_id': recordsets.id},
+                apply_tenant_criteria=apply_tenant_criteria,
+            )
 
             recordsets.obj_reset_changes(['records'])
 
@@ -610,7 +694,9 @@ class SQLAlchemyStorage(sqlalchemy_base.SQLAlchemy, storage_base.Storage):
                     context, criterion, tables.zones, tables.recordsets,
                     tables.records, limit=limit, marker=marker,
                     sort_key=sort_key, sort_dir=sort_dir,
-                    force_index=force_index)
+                    force_index=force_index,
+                    apply_tenant_criteria=apply_tenant_criteria,
+            )
 
             recordsets.total_count = tc
 
@@ -641,10 +727,7 @@ class SQLAlchemyStorage(sqlalchemy_base.SQLAlchemy, storage_base.Storage):
         return raw_rows
 
     def create_recordset(self, context, zone_id, recordset):
-        # Fetch the zone as we need the tenant_id
+        recordset.tenant_id = context.project_id
-        zone = self._find_zones(context, {'id': zone_id}, one=True)
-
-        recordset.tenant_id = zone.tenant_id
         recordset.zone_id = zone_id
 
         # Patch in the reverse_name column
@@ -687,17 +770,18 @@ class SQLAlchemyStorage(sqlalchemy_base.SQLAlchemy, storage_base.Storage):
 
         return raw_rows
 
-    def get_recordset(self, context, recordset_id):
-        return self._find_recordsets(context, {'id': recordset_id}, one=True)
-
     def find_recordsets(self, context, criterion=None, marker=None, limit=None,
-                        sort_key=None, sort_dir=None, force_index=False):
+                        sort_key=None, sort_dir=None, force_index=False,
-        return self._find_recordsets(context, criterion, marker=marker,
+                        apply_tenant_criteria=True):
-                                     sort_dir=sort_dir, sort_key=sort_key,
+        return self._find_recordsets(
-                                     limit=limit, force_index=force_index)
+            context, criterion, marker=marker, sort_dir=sort_dir,
+            sort_key=sort_key, limit=limit, force_index=force_index,
+            apply_tenant_criteria=apply_tenant_criteria)
 
-    def find_recordset(self, context, criterion):
+    def find_recordset(self, context, criterion, apply_tenant_criteria=True):
-        return self._find_recordsets(context, criterion, one=True)
+        return self._find_recordsets(
+            context, criterion, one=True,
+            apply_tenant_criteria=apply_tenant_criteria)
 
     def update_recordset(self, context, recordset):
         recordset = self._update(
@@ -779,11 +863,14 @@ class SQLAlchemyStorage(sqlalchemy_base.SQLAlchemy, storage_base.Storage):
 
     # Record Methods
     def _find_records(self, context, criterion, one=False, marker=None,
-                      limit=None, sort_key=None, sort_dir=None):
+                      limit=None, sort_key=None, sort_dir=None,
+                      apply_tenant_criteria=True):
         return self._find(
             context, tables.records, objects.Record, objects.RecordList,
             exceptions.RecordNotFound, criterion, one, marker, limit,
-            sort_key, sort_dir)
+            sort_key, sort_dir,
+            apply_tenant_criteria=apply_tenant_criteria,
+        )
 
     def _recalculate_record_hash(self, record):
         """
@@ -796,10 +883,7 @@ class SQLAlchemyStorage(sqlalchemy_base.SQLAlchemy, storage_base.Storage):
         return md5sum.hexdigest()
 
     def create_record(self, context, zone_id, recordset_id, record):
-        # Fetch the zone as we need the tenant_id
+        record.tenant_id = context.project_id
-        zone = self._find_zones(context, {'id': zone_id}, one=True)
-
-        record.tenant_id = zone.tenant_id
         record.zone_id = zone_id
         record.recordset_id = recordset_id
         record.hash = self._recalculate_record_hash(record)

designate/storage/impl_sqlalchemy/alembic/versions/b20189fd288e_shared_zone.py

@@ -0,0 +1,48 @@
+#    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.
+
+"""shared_zones
+
+Revision ID: b20189fd288e
+Revises: e5e2199ed76e
+Create Date: 2022-09-22 20:50:03.056609
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+from designate.sqlalchemy.types import UUID
+from designate import utils
+
+# revision identifiers, used by Alembic.
+revision = 'b20189fd288e'
+down_revision = 'e5e2199ed76e'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    meta = sa.MetaData()
+
+    op.create_table(
+        'shared_zones', meta,
+        sa.Column('id', UUID, default=utils.generate_uuid, primary_key=True),
+        sa.Column('created_at', sa.DateTime),
+        sa.Column('updated_at', sa.DateTime),
+        sa.Column('zone_id', UUID, nullable=False),
+        sa.Column('project_id', sa.String(36), nullable=False),
+        sa.Column('target_project_id', sa.String(36), nullable=False),
+
+        sa.UniqueConstraint('zone_id', 'project_id', 'target_project_id',
+                         name='unique_shared_zone'),
+        sa.ForeignKeyConstraint(['zone_id'], ['zones.id'], ondelete='CASCADE'),
+    )

designate/storage/impl_sqlalchemy/tables.py

@@ -183,6 +183,20 @@ zone_masters = Table('zone_masters', metadata,
     mysql_charset='utf8'
 )
 
+shared_zones = Table(
+    'shared_zones', metadata,
+    Column('id', UUID, default=utils.generate_uuid, primary_key=True),
+    Column('created_at', DateTime, default=lambda: timeutils.utcnow()),
+    Column('updated_at', DateTime, onupdate=lambda: timeutils.utcnow()),
+    Column('zone_id', UUID, nullable=False),
+    Column('project_id', String(36), nullable=False),
+    Column('target_project_id', String(36), nullable=False),
+
+    UniqueConstraint('zone_id', 'project_id', 'target_project_id',
+                     name='unique_shared_zone'),
+    ForeignKeyConstraint(('zone_id',), ['zones.id'], ondelete='CASCADE'),
+)
+
 recordsets = Table('recordsets', metadata,
     Column('id', UUID, default=utils.generate_uuid, primary_key=True),
     Column('version', Integer, default=1, nullable=False),

designate/tests/__init__.py

@@ -283,6 +283,14 @@ class TestCase(base.BaseTestCase):
          'port': 53},
     ]
 
+    shared_zone_fixtures = [
+        {
+            "target_project_id": "target_project_id",
+            "zone_id": None,
+            "project_id": "project_id",
+        }
+    ]
+
     zone_transfers_request_fixtures = [{
         "description": "Test Transfer",
     }, {
@@ -628,6 +636,13 @@ class TestCase(base.BaseTestCase):
         _values.update(values)
         return _values
 
+    def get_shared_zone_fixture(self, fixture=0, values=None):
+        values = values or {}
+
+        _values = copy.copy(self.shared_zone_fixtures[fixture])
+        _values.update(values)
+        return _values
+
     def update_service_status(self, **kwargs):
         context = kwargs.pop('context', self.admin_context)
         fixture = kwargs.pop('fixture', 0)
@@ -830,6 +845,16 @@ class TestCase(base.BaseTestCase):
 
         return zone_import
 
+    def share_zone(self, **kwargs):
+        context = kwargs.pop('context', self.admin_context)
+        fixture = kwargs.pop('fixture', 0)
+
+        values = self.get_shared_zone_fixture(fixture, values=kwargs)
+
+        return self.central_service.share_zone(
+            context, kwargs['zone_id'], objects.SharedZone.from_dict(values)
+        )
+
     def _ensure_interface(self, interface, implementation):
         for name in interface.__abstractmethods__:
             in_arginfo = inspect.getfullargspec(getattr(interface, name))

designate/tests/test_api/test_v2/test_shared_zones.py

@@ -0,0 +1,130 @@
+# Copyright 2020 Cloudification GmbH. 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 designate.tests.test_api.test_v2 import ApiV2TestCase
+
+
+class ApiV2SharedZonesTest(ApiV2TestCase):
+    def setUp(self):
+        super(ApiV2SharedZonesTest, self).setUp()
+
+        self.zone = self.create_zone()
+        self.target_project_id = '2'
+        self.endpoint_url = '/zones/{}/shares'
+
+    def _create_valid_shared_zone(self):
+        return self.client.post_json(
+            self.endpoint_url.format(self.zone.id),
+            {
+                'target_project_id': self.target_project_id,
+            }
+        )
+
+    def test_share_zone(self):
+        response = self._create_valid_shared_zone()
+
+        # Check the headers are what we expect
+        self.assertEqual(201, response.status_int)
+        self.assertEqual('application/json', response.content_type)
+
+        # Check the body structure is what we expect
+        self.assertIn('links', response.json)
+        self.assertIn('self', response.json['links'])
+
+        # Check the values returned are what we expect
+        self.assertIn('id', response.json)
+        self.assertIn('created_at', response.json)
+        self.assertEqual(
+            self.target_project_id,
+            response.json['target_project_id'])
+        self.assertEqual(
+            self.zone.id,
+            response.json['zone_id'])
+        self.assertIsNone(response.json['updated_at'])
+
+    def test_share_zone_with_no_target_id_no_zone_id(self):
+        self._assert_exception(
+            'invalid_uuid', 400, self.client.post_json,
+            self.endpoint_url.format(""), {"target_project_id": ""}
+        )
+
+    def test_share_zone_with_target_id_no_zone_id(self):
+        self._assert_exception(
+            'invalid_uuid', 400, self.client.post_json,
+            self.endpoint_url.format(""), {"target_project_id": "2"}
+        )
+
+    def test_share_zone_with_invalid_zone_id(self):
+        self._assert_exception(
+            'invalid_uuid', 400, self.client.post_json,
+            self.endpoint_url.format("invalid"), {"target_project_id": "2"}
+        )
+
+    def test_get_zone_share(self):
+        shared_zone = self._create_valid_shared_zone()
+
+        response = self.client.get(
+            '{}/{}'.format(self.endpoint_url.format(self.zone.id),
+                           shared_zone.json['id'])
+        )
+
+        # Check the headers are what we expect
+        self.assertEqual(200, response.status_int)
+        self.assertEqual('application/json', response.content_type)
+
+        # Check the body structure is what we expect
+        self.assertIn('links', response.json)
+        self.assertIn('self', response.json['links'])
+
+        # Check the values returned are what we expect
+        self.assertIn('id', response.json)
+        self.assertIn('created_at', response.json)
+        self.assertEqual(
+            self.target_project_id,
+            response.json['target_project_id'])
+        self.assertEqual(
+            self.zone.id,
+            response.json['zone_id'])
+        self.assertIn('updated_at', response.json)
+
+    def test_list_zone_shares(self):
+        response = self.client.get(self.endpoint_url.format(self.zone.id))
+
+        # Check the headers are what we expect
+        self.assertEqual(200, response.status_int)
+        self.assertEqual('application/json', response.content_type)
+
+        # Check the body structure is what we expect
+        self.assertIn('shared_zones', response.json)
+        self.assertIn('links', response.json)
+        self.assertIn('self', response.json['links'])
+
+        # We should start with 0 zone shars
+        self.assertEqual(0, len(response.json['shared_zones']))
+
+        self._create_valid_shared_zone()
+
+        data = self.client.get(self.endpoint_url.format(self.zone.id))
+
+        self.assertEqual(1, len(data.json['shared_zones']))
+
+    def test_delete_zone_share(self):
+        shared_zone = self._create_valid_shared_zone()
+
+        response = self.client.delete(
+            '{}/{}'.format(self.endpoint_url.format(self.zone.id),
+                           shared_zone.json['id'])
+        )
+
+        # Check the headers are what we expect
+        self.assertEqual(204, response.status_int)

designate/tests/test_central/test_service.py

@@ -24,6 +24,7 @@ import random
 from unittest import mock
 
 from oslo_config import cfg
+from oslo_config import fixture as cfg_fixture
 from oslo_db import exception as db_exception
 from oslo_log import log as logging
 from oslo_messaging.notify import notifier
@@ -421,8 +422,8 @@ class CentralServiceTest(CentralTestCase):
         admin_context = self.get_admin_context()
         admin_context.all_tenants = True
 
-        tenant_one_context = self.get_context(project_id=1)
+        tenant_one_context = self.get_context(project_id='1')
-        tenant_two_context = self.get_context(project_id=2)
+        tenant_two_context = self.get_context(project_id='2')
 
         # in the beginning, there should be nothing
         tenants = self.central_service.count_tenants(admin_context)
@@ -719,7 +720,7 @@ class CentralServiceTest(CentralTestCase):
         self.policy({'use_low_ttl': '!'})
         self.config(min_ttl=100,
                     group='service:central')
-        context = self.get_context(project_id=1)
+        context = self.get_context(project_id='1')
 
         values = self.get_zone_fixture(fixture=1)
         values['ttl'] = 5
@@ -796,8 +797,8 @@ class CentralServiceTest(CentralTestCase):
         admin_context = self.get_admin_context()
         admin_context.all_tenants = True
 
-        tenant_one_context = self.get_context(project_id=1)
+        tenant_one_context = self.get_context(project_id='1')
-        tenant_two_context = self.get_context(project_id=2)
+        tenant_two_context = self.get_context(project_id='2')
 
         # Ensure we have no zones to start with.
         zones = self.central_service.find_zones(admin_context)
@@ -822,7 +823,7 @@ class CentralServiceTest(CentralTestCase):
 
     def test_get_zone(self):
         # Create a zone
-        zone_name = '%d.example.com.' % random.randint(10, 1000)
+        zone_name = '%d.example.com.' % random.randint(10, 10000)
         expected_zone = self.create_zone(name=zone_name)
 
         # Retrieve it, and ensure it's the same
@@ -833,6 +834,42 @@ class CentralServiceTest(CentralTestCase):
         self.assertEqual(expected_zone['name'], zone['name'])
         self.assertEqual(expected_zone['email'], zone['email'])
 
+    def test_get_zone_not_owner_not_shared(self):
+        # Create a zone
+        zone_name = '%d.example.com.' % random.randint(10, 10000)
+        expected_zone = self.create_zone(name=zone_name)
+
+        context = self.get_context(project_id='fake')
+
+        with mock.patch.object(self.central_service.storage,
+                               'is_zone_shared_with_project',
+                               return_value=False):
+            # Make sure random projects can't get the zone
+            exc = self.assertRaises(rpc_dispatcher.ExpectedException,
+                                    self.central_service.get_zone,
+                                    context, expected_zone['id'],
+                                    apply_tenant_criteria=False)
+            self.assertEqual(exceptions.ZoneNotFound, exc.exc_info[0])
+
+    def test_get_zone_not_owner_shared(self):
+        # Create a zone
+        zone_name = '%d.example.com.' % random.randint(10, 10000)
+        expected_zone = self.create_zone(name=zone_name)
+
+        context = self.get_context(project_id='fake')
+
+        with mock.patch.object(self.central_service.storage,
+                               'is_zone_shared_with_project',
+                               return_value=True):
+
+            # Retrieve it, and ensure it's the same
+            zone = self.central_service.get_zone(context, expected_zone['id'],
+                                                 apply_tenant_criteria=False)
+
+            self.assertEqual(expected_zone['id'], zone['id'])
+            self.assertEqual(expected_zone['name'], zone['name'])
+            self.assertEqual(expected_zone['email'], zone['email'])
+
     def test_get_zone_servers(self):
         # Create a zone
         zone = self.create_zone()
@@ -985,6 +1022,51 @@ class CentralServiceTest(CentralTestCase):
         self.assertIsInstance(notified_zone, objects.Zone)
         self.assertEqual(deleted_zone.id, notified_zone.id)
 
+    @mock.patch.object(notifier.Notifier, "info")
+    def test_delete_zone_shared_no_delete_shares(self, mock_notifier):
+        # Create a zone
+        zone = self.create_zone()
+
+        # Share the zone
+        self.share_zone(context=self.admin_context, zone_id=zone.id)
+
+        mock_notifier.reset_mock()
+
+        # Delete the zone
+        self.assertRaises(exceptions.ZoneShared,
+                          self.central_service.delete_zone,
+                          self.admin_context, zone['id'])
+
+    @mock.patch.object(notifier.Notifier, "info")
+    def test_delete_zone_shared_delete_shares(self, mock_notifier):
+        context = self.get_admin_context(delete_shares=True)
+
+        # Create a zone
+        zone = self.create_zone(context=context)
+
+        # Share the zone
+        self.share_zone(context=context, zone_id=zone.id)
+
+        mock_notifier.reset_mock()
+
+        # Delete the zone
+        self.central_service.delete_zone(context, zone.id)
+
+        # Fetch the zone
+        deleted_zone = self.central_service.get_zone(context, zone['id'])
+
+        # Ensure the zone is marked for deletion
+        self.assertEqual(zone.id, deleted_zone.id)
+        self.assertEqual(zone.name, deleted_zone.name)
+        self.assertEqual(zone.email, deleted_zone.email)
+        self.assertEqual('PENDING', deleted_zone.status)
+        self.assertEqual(zone.tenant_id, deleted_zone.tenant_id)
+        self.assertEqual(zone.parent_zone_id,
+                         deleted_zone.parent_zone_id)
+        self.assertEqual('DELETE', deleted_zone.action)
+        self.assertEqual(zone.serial, deleted_zone.serial)
+        self.assertEqual(zone.pool_id, deleted_zone.pool_id)
+
     def test_delete_parent_zone(self):
         # Create the Parent Zone using fixture 0
         parent_zone = self.create_zone(fixture=0)
@@ -1391,6 +1473,69 @@ class CentralServiceTest(CentralTestCase):
         # in the recordset
         self.assertEqual(original_serial, new_serial)
 
+    def test_create_recordset_shared_zone(self):
+        zone = self.create_zone()
+        original_serial = zone.serial
+
+        # Create the Object
+        recordset = objects.RecordSet(name='www.%s' % zone.name, type='A')
+
+        context = self.get_context(project_id='1')
+        self.share_zone(context=self.admin_context, zone_id=zone.id,
+                        target_project_id='1')
+
+        # Persist the Object
+        recordset = self.central_service.create_recordset(
+            context, zone.id, recordset=recordset)
+
+        # Get the zone again to check if serial increased
+        updated_zone = self.central_service.get_zone(self.admin_context,
+                                                     zone.id)
+        new_serial = updated_zone.serial
+
+        # Ensure all values have been set correctly
+        self.assertIsNotNone(recordset.id)
+        self.assertEqual('www.%s' % zone.name, recordset.name)
+        self.assertEqual('A', recordset.type)
+
+        self.assertIsNotNone(recordset.records)
+        # The serial number does not get updated is there are no records
+        # in the recordset
+        self.assertEqual(original_serial, new_serial)
+
+    def test_create_recordset_shared_zone_new_policy_defaults(self):
+        zone = self.create_zone()
+        original_serial = zone.serial
+
+        # Create the Object
+        recordset = objects.RecordSet(name='www.%s' % zone.name, type='A')
+
+        self.useFixture(cfg_fixture.Config(cfg.CONF))
+        cfg.CONF.set_override('enforce_new_defaults', True, 'oslo_policy')
+        context = self.get_context(project_id='1', roles=['member', 'reader'])
+
+        self.share_zone(context=self.admin_context, zone_id=zone.id,
+                        target_project_id='1')
+
+        # Persist the Object
+        recordset = self.central_service.create_recordset(
+            context, zone.id, recordset=recordset)
+
+        # Get the zone again to check if serial increased
+        updated_zone = self.central_service.get_zone(self.admin_context,
+                                                     zone.id)
+        new_serial = updated_zone.serial
+
+        # Ensure all values have been set correctly
+        self.assertIsNotNone(recordset.id)
+        self.assertEqual('www.%s' % zone.name, recordset.name)
+        self.assertEqual('A', recordset.type)
+
+        self.assertIsNotNone(recordset.records)
+        # The serial number does not get updated is there are no records
+        # in the recordset
+        self.assertEqual(original_serial, new_serial)
+
     def test_create_recordset_with_records(self):
         zone = self.create_zone()
         original_serial = zone.serial
@@ -1573,6 +1718,24 @@ class CentralServiceTest(CentralTestCase):
 
         self.assertEqual(exceptions.RecordSetNotFound, exc.exc_info[0])
 
+    def test_get_recordset_shared_zone(self):
+        zone = self.create_zone()
+
+        context = self.get_context(project_id='1')
+        self.share_zone(context=self.admin_context, zone_id=zone.id,
+                        target_project_id='1')
+
+        # Create a recordset
+        expected = self.create_recordset(zone)
+
+        # Retrieve it, and ensure it's the same
+        recordset = self.central_service.get_recordset(
+            context, zone['id'], expected['id'])
+
+        self.assertEqual(expected['id'], recordset['id'])
+        self.assertEqual(expected['name'], recordset['name'])
+        self.assertEqual(expected['type'], recordset['type'])
+
     def test_find_recordsets(self):
         zone = self.create_zone()
 
@@ -1606,6 +1769,41 @@ class CentralServiceTest(CentralTestCase):
         self.assertEqual('www.%s' % zone['name'], recordsets[2]['name'])
         self.assertEqual('mail.%s' % zone['name'], recordsets[3]['name'])
 
+    def test_find_recordsets_shared_zone(self):
+        zone = self.create_zone()
+
+        context = self.get_context(project_id='1')
+        self.share_zone(context=self.admin_context, zone_id=zone.id,
+                        target_project_id='1')
+
+        criterion = {'zone_id': zone['id']}
+
+        # Create a single recordset (using default values)
+        self.create_recordset(zone, name='www.%s' % zone['name'])
+
+        # Ensure we can retrieve the newly created recordset
+        recordsets = self.central_service.find_recordsets(context, criterion)
+
+        self.assertEqual(3, len(recordsets))
+        self.assertEqual('www.%s' % zone['name'], recordsets[2]['name'])
+
+    def test_find_recordsets_not_shared_zone(self):
+        zone = self.create_zone()
+
+        context = self.get_context(project_id='2')
+
+        criterion = {'zone_id': zone['id']}
+
+        # Create a single recordset (using default values)
+        self.create_recordset(zone, name='www.%s' % zone['name'])
+
+        # Ensure we can retrieve the newly created recordset
+        exc = self.assertRaises(rpc_dispatcher.ExpectedException,
+                                self.central_service.find_recordsets,
+                                context, criterion)
+
+        self.assertEqual(exceptions.ZoneNotFound, exc.exc_info[0])
+
     def test_find_recordset(self):
         zone = self.create_zone()
 
@@ -1856,6 +2054,37 @@ class CentralServiceTest(CentralTestCase):
         self.assertRaises(ovo_exc.ReadOnlyFieldError, setattr,
                           recordset, 'type', cname_recordset.type)
 
+    def test_update_recordset_shared_zone(self):
+        # Create a zone
+        zone = self.create_zone()
+        original_serial = zone.serial
+
+        context = self.get_context(project_id='1')
+        self.share_zone(context=self.admin_context, zone_id=zone.id,
+                        target_project_id='1')
+
+        # Create a recordset
+        recordset = self.create_recordset(zone, context=context)
+
+        # Update the recordset
+        recordset.ttl = 1800
+
+        # Perform the update
+        self.central_service.update_recordset(context, recordset)
+
+        # Get zone again to verify that serial number was updated
+        updated_zone = self.central_service.get_zone(self.admin_context,
+                                                     zone.id)
+        new_serial = updated_zone.serial
+
+        # Fetch the resource again
+        recordset = self.central_service.get_recordset(
+            self.admin_context, recordset.zone_id, recordset.id)
+
+        # Ensure the new value took
+        self.assertEqual(1800, recordset.ttl)
+        self.assertThat(new_serial, GreaterThan(original_serial))
+
     def test_delete_recordset(self):
         zone = self.create_zone()
         original_serial = zone.serial
@@ -3078,14 +3307,14 @@ class CentralServiceTest(CentralTestCase):
         self.assertEqual(zt_request.key, retrived_zt.key)
 
     def test_get_zone_transfer_request_scoped(self):
-        tenant_1_context = self.get_context(project_id=1)
+        tenant_1_context = self.get_context(project_id='1')
-        tenant_2_context = self.get_context(project_id=2)
+        tenant_2_context = self.get_context(project_id='2')
-        tenant_3_context = self.get_context(project_id=3)
+        tenant_3_context = self.get_context(project_id='3')
         zone = self.create_zone(context=tenant_1_context)
         zt_request = self.create_zone_transfer_request(
             zone,
             context=tenant_1_context,
-            target_tenant_id=2)
+            target_tenant_id='2')
 
         self.central_service.get_zone_transfer_request(
             tenant_2_context, zt_request.id)
@@ -3129,8 +3358,8 @@ class CentralServiceTest(CentralTestCase):
                          exc.exc_info[0])
 
     def test_create_zone_transfer_accept(self):
-        tenant_1_context = self.get_context(project_id=1)
+        tenant_1_context = self.get_context(project_id='1')
-        tenant_2_context = self.get_context(project_id=2)
+        tenant_2_context = self.get_context(project_id="2")
         admin_context = self.get_admin_context()
         admin_context.all_tenants = True
 
@@ -3179,8 +3408,8 @@ class CentralServiceTest(CentralTestCase):
             'COMPLETE', result['zt_request'].status)
 
     def test_create_zone_transfer_accept_scoped(self):
-        tenant_1_context = self.get_context(project_id=1)
+        tenant_1_context = self.get_context(project_id='1')
-        tenant_2_context = self.get_context(project_id=2)
+        tenant_2_context = self.get_context(project_id="2")
         admin_context = self.get_admin_context()
         admin_context.all_tenants = True
 
@@ -3231,8 +3460,8 @@ class CentralServiceTest(CentralTestCase):
             'COMPLETE', result['zt_request'].status)
 
     def test_create_zone_transfer_accept_failed_key(self):
-        tenant_1_context = self.get_context(project_id=1)
+        tenant_1_context = self.get_context(project_id='1')
-        tenant_2_context = self.get_context(project_id=2)
+        tenant_2_context = self.get_context(project_id="2")
         admin_context = self.get_admin_context()
         admin_context.all_tenants = True
 
@@ -3241,7 +3470,7 @@ class CentralServiceTest(CentralTestCase):
         zone_transfer_request = self.create_zone_transfer_request(
             zone,
             context=tenant_1_context,
-            target_tenant_id=2)
+            target_tenant_id="2")
 
         zone_transfer_accept = objects.ZoneTransferAccept()
         zone_transfer_accept.zone_transfer_request_id =\
@@ -3258,8 +3487,8 @@ class CentralServiceTest(CentralTestCase):
         self.assertEqual(exceptions.IncorrectZoneTransferKey, exc.exc_info[0])
 
     def test_create_zone_tarnsfer_accept_out_of_tenant_scope(self):
-        tenant_1_context = self.get_context(project_id=1)
+        tenant_1_context = self.get_context(project_id='1')
-        tenant_3_context = self.get_context(project_id=3)
+        tenant_3_context = self.get_context(project_id="3")
         admin_context = self.get_admin_context()
         admin_context.all_tenants = True
 
@@ -3268,7 +3497,7 @@ class CentralServiceTest(CentralTestCase):
         zone_transfer_request = self.create_zone_transfer_request(
             zone,
             context=tenant_1_context,
-            target_tenant_id=2)
+            target_tenant_id="2")
 
         zone_transfer_accept = objects.ZoneTransferAccept()
         zone_transfer_accept.zone_transfer_request_id =\
@@ -3532,3 +3761,174 @@ class CentralServiceTest(CentralTestCase):
                                 context, zone_import['id'])
 
         self.assertEqual(exceptions.ZoneImportNotFound, exc.exc_info[0])
+
+    def test_share_zone(self):
+        # Create a Shared Zone
+        context = self.get_context(project_id='1')
+        zone = self.create_zone(context=context)
+        shared_zone = self.share_zone(context=context, zone_id=zone.id)
+
+        # Ensure all values have been set correctly
+        self.assertIsNotNone(shared_zone['id'])
+        self.assertEqual('target_project_id', shared_zone.target_project_id)
+        self.assertEqual(context.project_id, shared_zone.project_id)
+        self.assertEqual(zone.id, shared_zone.zone_id)
+
+    def test_share_zone_new_policy_defaults(self):
+        # Configure designate for enforcing the new policy defaults
+        self.useFixture(cfg_fixture.Config(cfg.CONF))
+        cfg.CONF.set_override('enforce_new_defaults', True, 'oslo_policy')
+        context = self.get_context(project_id='1', roles=['member', 'reader'])
+
+        # Create a Shared Zone
+        zone = self.create_zone(context=context)
+        shared_zone = self.share_zone(context=context, zone_id=zone.id)
+
+        # Ensure all values have been set correctly
+        self.assertIsNotNone(shared_zone['id'])
+        self.assertEqual('target_project_id', shared_zone.target_project_id)
+        self.assertEqual(context.project_id, shared_zone.project_id)
+        self.assertEqual(zone.id, shared_zone.zone_id)
+
+    def test_unshare_zone(self):
+        context = self.get_context(project_id='1')
+        zone = self.create_zone(context=context)
+        shared_zone = self.share_zone(context=context, zone_id=zone.id)
+
+        new_shared_zone_obj = self.central_service.unshare_zone(
+            context, zone.id, shared_zone.id
+        )
+
+        self.assertEqual(shared_zone.id, new_shared_zone_obj.id)
+        self.assertEqual(shared_zone.target_project_id,
+                         new_shared_zone_obj.target_project_id)
+        self.assertEqual(shared_zone.project_id,
+                         new_shared_zone_obj.project_id)
+
+    def test_unshare_zone_new_policy_defaults(self):
+        # Configure designate for enforcing the new policy defaults
+        self.useFixture(cfg_fixture.Config(cfg.CONF))
+        cfg.CONF.set_override('enforce_new_defaults', True, 'oslo_policy')
+        context = self.get_context(project_id='1', roles=['member', 'reader'])
+
+        # Create a Shared Zone
+        zone = self.create_zone(context=context)
+        shared_zone = self.share_zone(context=context, zone_id=zone.id)
+
+        new_shared_zone_obj = self.central_service.unshare_zone(
+            context, zone.id, shared_zone.id
+        )
+
+        self.assertEqual(shared_zone.id, new_shared_zone_obj.id)
+        self.assertEqual(shared_zone.target_project_id,
+                         new_shared_zone_obj.target_project_id)
+        self.assertEqual(shared_zone.project_id,
+                         new_shared_zone_obj.project_id)
+
+    def test_unshare_zone_with_child_objects(self):
+        context = self.get_context(project_id='1')
+        zone = self.create_zone(context=context)
+        shared_zone = self.share_zone(context=context, zone_id=zone.id)
+
+        with mock.patch.object(self.central_service.storage,
+                               'count_zones', return_value=1):
+            exc = self.assertRaises(rpc_dispatcher.ExpectedException,
+                                    self.central_service.unshare_zone,
+                                    context, zone.id, shared_zone.id)
+
+            self.assertEqual(exceptions.SharedZoneHasSubZone, exc.exc_info[0])
+
+        with mock.patch.object(self.central_service.storage,
+                               'count_recordsets', return_value=1):
+            exc = self.assertRaises(rpc_dispatcher.ExpectedException,
+                                    self.central_service.unshare_zone,
+                                    context, zone.id, shared_zone.id)
+
+            self.assertEqual(
+                exceptions.SharedZoneHasRecordSets,
+                exc.exc_info[0]
+            )
+
+    def test_find_shared_zones(self):
+        context = self.get_context(project_id='1')
+        zone = self.create_zone(context=context)
+
+        # Ensure we have no shared zones to start with.
+        shared_zones = self.central_service.find_shared_zones(context,
+            criterion={'zone_id': zone.id})
+        self.assertEqual(0, len(shared_zones))
+
+        # Create a first shared_zone
+        shared_zone = self.share_zone(context=context, zone_id=zone.id)
+
+        # Ensure we can retrieve the newly created shared_zone
+        shared_zones = self.central_service.find_shared_zones(context,
+            criterion={'zone_id': zone.id})
+        self.assertEqual(1, len(shared_zones))
+
+        # Ensure we can retrieve the newly created shared_zone no criteria
+        shared_zones = self.central_service.find_shared_zones(context)
+        self.assertEqual(1, len(shared_zones))
+
+        # Create a second shared_zone
+        second_shared_zone = self.share_zone(
+            context=context, zone_id=zone.id, target_project_id="second_tenant"
+        )
+
+        # Ensure we can retrieve both shared_zones
+        shared_zones = self.central_service.find_shared_zones(context,
+            criterion={'zone_id': zone.id})
+
+        self.assertEqual(2, len(shared_zones))
+        self.assertEqual(zone.id, shared_zones[0].zone_id)
+        self.assertEqual(shared_zone.id, shared_zones[0].id)
+        self.assertEqual(zone.id, shared_zones[1].zone_id)
+        self.assertEqual(second_shared_zone.id, shared_zones[1].id)
+
+    def test_find_shared_zones_new_policy_defaults(self):
+        # Configure designate for enforcing the new policy defaults
+        context = self.get_context(project_id='1', roles=['member', 'reader'])
+
+        zone = self.create_zone(context=context)
+
+        # Create a first shared_zone
+        shared_zone = self.share_zone(context=context, zone_id=zone.id)
+
+        # Ensure we can retrieve the newly created shared_zone
+        shared_zones = self.central_service.find_shared_zones(context,
+            criterion={'zone_id': zone.id})
+        self.assertEqual(1, len(shared_zones))
+
+        # Create a second shared_zone
+        second_shared_zone = self.share_zone(
+            context=context, zone_id=zone.id, target_project_id="second_tenant"
+        )
+
+        self.useFixture(cfg_fixture.Config(cfg.CONF))
+        cfg.CONF.set_override('enforce_new_defaults', True, 'oslo_policy')
+
+        # Ensure we can retrieve both shared_zones
+        shared_zones = self.central_service.find_shared_zones(context,
+            criterion={'zone_id': zone.id})
+
+        self.assertEqual(2, len(shared_zones))
+        self.assertEqual(zone.id, shared_zones[0].zone_id)
+        self.assertEqual(shared_zone.id, shared_zones[0].id)
+        self.assertEqual(zone.id, shared_zones[1].zone_id)
+        self.assertEqual(second_shared_zone.id, shared_zones[1].id)
+
+    def test_get_shared_zone(self):
+        context = self.get_context(project_id='1')
+        zone = self.create_zone(context=context)
+
+        shared_zone = self.share_zone(context=context, zone_id=zone.id)
+
+        retrived_shared_zone = self.central_service.get_shared_zone(
+            context, zone.id, shared_zone.id)
+
+        self.assertEqual(zone.id, retrived_shared_zone.zone_id)
+        self.assertEqual(shared_zone.id, retrived_shared_zone.id)
+        self.assertEqual(shared_zone.target_project_id,
+                         retrived_shared_zone.target_project_id)
+        self.assertEqual(shared_zone.project_id,
+                         retrived_shared_zone.project_id)

designate/tests/test_storage/__init__.py

@@ -1049,41 +1049,6 @@ class StorageTestCase(object):
             self.assertNotIn(record, records)
             records.append(record)
 
-    def test_get_recordset(self):
-        zone = self.create_zone()
-        expected = self.create_recordset(zone)
-
-        actual = self.storage.get_recordset(self.admin_context, expected['id'])
-
-        self.assertEqual(expected['name'], actual['name'])
-        self.assertEqual(expected['type'], actual['type'])
-
-    def test_get_recordset_with_records(self):
-        zone = self.create_zone()
-
-        records = [
-            objects.Record.from_dict(self.get_record_fixture('A', fixture=0)),
-            objects.Record.from_dict(self.get_record_fixture('A', fixture=1))
-        ]
-        recordset = self.create_recordset(zone, records=records)
-
-        # Fetch the RecordSet again
-        recordset = self.storage.get_recordset(
-            self.admin_context, recordset.id)
-
-        # Ensure recordset.records is a RecordList instance
-        self.assertIsInstance(recordset.records, objects.RecordList)
-
-        # Ensure two Records are attached to the RecordSet correctly
-        self.assertEqual(2, len(recordset.records))
-        self.assertIsInstance(recordset.records[0], objects.Record)
-        self.assertIsInstance(recordset.records[1], objects.Record)
-
-    def test_get_recordset_missing(self):
-        with testtools.ExpectedException(exceptions.RecordSetNotFound):
-            uuid = 'caf771fc-6b05-4891-bee1-c2a48621f57b'
-            self.storage.get_recordset(self.admin_context, uuid)
-
     def test_find_recordset_criterion(self):
         zone = self.create_zone()
         expected = self.create_recordset(zone)
@@ -1189,8 +1154,8 @@ class StorageTestCase(object):
         self.storage.update_recordset(self.admin_context, recordset)
 
         # Fetch the RecordSet again
-        recordset = self.storage.get_recordset(
+        recordset = self.storage.find_recordset(self.admin_context,
-            self.admin_context, recordset.id)
+                                                {'id': recordset.id})
 
         # Ensure two Records are attached to the RecordSet correctly
         self.assertEqual(2, len(recordset.records))
@@ -1212,8 +1177,8 @@ class StorageTestCase(object):
         recordset = self.create_recordset(zone, records=records)
 
         # Fetch the RecordSet again
-        recordset = self.storage.get_recordset(
+        recordset = self.storage.find_recordset(self.admin_context,
-            self.admin_context, recordset.id)
+                                                {'id': recordset.id})
 
         # Remove one of the Records
         recordset.records.pop(0)
@@ -1225,8 +1190,8 @@ class StorageTestCase(object):
         self.storage.update_recordset(self.admin_context, recordset)
 
         # Fetch the RecordSet again
-        recordset = self.storage.get_recordset(
+        recordset = self.storage.find_recordset(self.admin_context,
-            self.admin_context, recordset.id)
+                                                {'id': recordset.id})
 
         # Ensure only one Record is attached to the RecordSet
         self.assertEqual(1, len(recordset.records))
@@ -1243,8 +1208,8 @@ class StorageTestCase(object):
         recordset = self.create_recordset(zone, records=records)
 
         # Fetch the RecordSet again
-        recordset = self.storage.get_recordset(
+        recordset = self.storage.find_recordset(self.admin_context,
-            self.admin_context, recordset.id)
+                                                {'id': recordset.id})
 
         # Update one of the Records
         updated_record_id = recordset.records[0].id
@@ -1254,8 +1219,8 @@ class StorageTestCase(object):
         self.storage.update_recordset(self.admin_context, recordset)
 
         # Fetch the RecordSet again
-        recordset = self.storage.get_recordset(
+        recordset = self.storage.find_recordset(self.admin_context,
-            self.admin_context, recordset.id)
+                                                {'id': recordset.id})
 
         # Ensure the Record has been updated
         for record in recordset.records:
@@ -1276,7 +1241,8 @@ class StorageTestCase(object):
         self.storage.delete_recordset(self.admin_context, recordset['id'])
 
         with testtools.ExpectedException(exceptions.RecordSetNotFound):
-            self.storage.get_recordset(self.admin_context, recordset['id'])
+            self.storage.find_recordset(self.admin_context,
+                                        criterion={'id': recordset['id']})
 
     def test_delete_recordset_missing(self):
         with testtools.ExpectedException(exceptions.RecordSetNotFound):
@@ -3033,8 +2999,8 @@ class StorageTestCase(object):
 
         saved_zone = self.storage.get_zone(
             admin_context, zone.id)
-        saved_recordset = self.storage.get_recordset(
+        saved_recordset = self.storage.find_recordset(
-            admin_context, recordset.id)
+            admin_context, criterion={'id': recordset.id})
         saved_record = self.storage.get_record(
             admin_context, record.id)
 

designate/tests/test_storage/test_sqlalchemy.py

@@ -43,6 +43,7 @@ class SqlalchemyStorageTest(StorageTestCase, TestCase):
             'records',
             'recordsets',
             'service_statuses',
+            'shared_zones',
             'tlds',
             'tsigkeys',
             'zone_attributes',

designate/tests/unit/api/test_middleware.py

@@ -163,3 +163,32 @@ class KeystoneContextMiddlewareTest(oslotest.base.BaseTestCase):
 
         self.app(self.request)
         self.assertFalse(self.ctxt.hard_delete)
+
+    def test_delete_shares_not_set(self):
+        self.request.headers.update({
+            'X-Tenant-ID': 'TenantID',
+            'X-Roles': 'admin',
+        })
+
+        self.app(self.request)
+        self.assertFalse(self.ctxt.delete_shares)
+
+    def test_delete_shares_false(self):
+        self.request.headers.update({
+            'X-Tenant-ID': 'TenantID',
+            'X-Roles': 'admin',
+            'X-Designate-Delete-Shares': 'false'
+        })
+
+        self.app(self.request)
+        self.assertFalse(self.ctxt.delete_shares)
+
+    def test_delete_shares_true(self):
+        self.request.headers.update({
+            'X-Tenant-ID': 'TenantID',
+            'X-Roles': 'admin',
+            'X-Designate-Delete-Shares': 'True'
+        })
+
+        self.app(self.request)
+        self.assertTrue(self.ctxt.delete_shares)

designate/tests/unit/api/test_version.py

@@ -55,7 +55,7 @@ class TestApiVersion(oslotest.base.BaseTestCase):
         self.assertEqual(200, response.status_int)
         self.assertEqual('application/json', response.content_type)
 
-        self.assertEqual(2, len(response.json['versions']))
+        self.assertEqual(3, len(response.json['versions']))
         self.assertEqual(
             'http://127.0.0.2:9001/v2',
             response.json['versions'][0]['links'][0]['href']
@@ -71,7 +71,7 @@ class TestApiVersion(oslotest.base.BaseTestCase):
         self.assertEqual(200, response.status_int)
         self.assertEqual('application/json', response.content_type)
 
-        self.assertEqual(2, len(response.json['versions']))
+        self.assertEqual(3, len(response.json['versions']))
         self.assertEqual(
             'http://localhost/v2',
             response.json['versions'][0]['links'][0]['href']

designate/tests/unit/test_central/test_basic.py

@@ -147,6 +147,7 @@ class Mockzone(object):
     ttl = 1
     type = "PRIMARY"
     serial = 123
+    shared = False
 
     def obj_attr_is_set(self, n):
         if n == 'recordsets':
@@ -232,6 +233,7 @@ class CentralBasic(TestCase):
         attrs = {
             'count_zones.return_value': 0,
             'find_zone.return_value': Mockzone(),
+            'get_zone.return_value': Mockzone(),
             'get_pool.return_value': MockPool(),
             'find_pools.return_value': pool_list,
         }
@@ -264,7 +266,8 @@ class CentralBasic(TestCase):
             'sudo',
             'abandon',
             'all_tenants',
-            'hard_delete'
+            'hard_delete',
+            'project_id'
         ])
 
         self.service = Service()
@@ -534,6 +537,7 @@ class CentralZoneTestCase(CentralBasic):
     recordset__id_2 = 'dc85d9b0-1e9d-4e99-aede-a06664f1af2e'
     recordset__id_3 = '2a94a9fe-30d1-4a15-9071-0bb21996d971'
     zone_export__id = 'e887597f-9697-47dd-a202-7a2711f8669c'
+    zone_shared = False
 
     def setUp(self):
         super(CentralZoneTestCase, self).setUp()
@@ -889,6 +893,7 @@ class CentralZoneTestCase(CentralBasic):
         self.service.storage.get_zone.return_value = RoObject(
             name='foo',
             tenant_id='2',
+            shared=self.zone_shared,
         )
         self.service.get_zone(self.context,
                               CentralZoneTestCase.zone__id)
@@ -927,6 +932,7 @@ class CentralZoneTestCase(CentralBasic):
         self.service.storage.get_zone.return_value = RoObject(
             name='foo',
             tenant_id='2',
+            shared=self.zone_shared,
         )
         self.service.storage.count_zones.return_value = 2
 
@@ -943,7 +949,8 @@ class CentralZoneTestCase(CentralBasic):
         self.service.storage.get_zone.return_value = RoObject(
             name='foo',
             tenant_id='2',
-            id=CentralZoneTestCase.zone__id_2
+            id=CentralZoneTestCase.zone__id_2,
+            shared=self.zone_shared,
         )
         designate.central.service.policy = mock.NonCallableMock(spec_set=[
             'reset',
@@ -967,6 +974,7 @@ class CentralZoneTestCase(CentralBasic):
         self.service.storage.get_zone.return_value = RoObject(
             name='foo',
             tenant_id='2',
+            shared=self.zone_shared,
         )
         self.service._delete_zone_in_storage = mock.Mock(
             return_value=RoObject(
@@ -995,6 +1003,7 @@ class CentralZoneTestCase(CentralBasic):
         self.service.storage.get_zone.return_value = RoObject(
             name='foo',
             tenant_id='2',
+            shared=False
         )
         self.service._delete_zone_in_storage = mock.Mock(
             return_value=RoObject(
@@ -1116,10 +1125,10 @@ class CentralZoneTestCase(CentralBasic):
         self.assertEqual(exceptions.ReportNotFound, exc.exc_info[0])
 
     def test_get_recordset_not_found(self):
-        self.service.storage.get_zone.return_value = RoObject(
+        zone = Mockzone()
-            id=CentralZoneTestCase.zone__id,
+        zone.id = CentralZoneTestCase.zone__id
-        )
+        self.service.storage.get_zone.return_value = zone
-        self.service.storage.get_recordset.return_value = RoObject(
+        self.service.storage.find_recordset.return_value = RoObject(
             zone_id=CentralZoneTestCase.zone__id_2
         )
 
@@ -1136,13 +1145,16 @@ class CentralZoneTestCase(CentralBasic):
             id=CentralZoneTestCase.zone__id_2,
             name='example.org.',
             tenant_id='2',
+            shared=self.zone_shared,
         )
-        self.service.storage.get_recordset.return_value = (
+        recordset = objects.RecordSet(
-            objects.RecordSet(
+            zone_id=CentralZoneTestCase.zone__id_2,
-                zone_id=CentralZoneTestCase.zone__id_2,
+            zone_name='example.org.',
-                zone_name='example.org.',
+            id=CentralZoneTestCase.recordset__id
-                id=CentralZoneTestCase.recordset__id
+        )
-            ))
+
+        self.service.storage.find_recordset.return_value = recordset
+
         self.service.get_recordset(
             self.context,
             CentralZoneTestCase.zone__id_2,
@@ -1157,6 +1169,41 @@ class CentralZoneTestCase(CentralBasic):
         self.assertEqual({
             'zone_id': CentralZoneTestCase.zone__id_2,
             'zone_name': 'example.org.',
+            'zone_shared': self.zone_shared,
+            'recordset_id': CentralZoneTestCase.recordset__id,
+            'project_id': '2'}, target)
+
+    def test_get_recordset_no_zone_id(self):
+        self.service.storage.get_zone.return_value = RoObject(
+            id=CentralZoneTestCase.zone__id_2,
+            name='example.org.',
+            tenant_id='2',
+            shared=self.zone_shared,
+        )
+        recordset = objects.RecordSet(
+            zone_id=CentralZoneTestCase.zone__id_2,
+            zone_name='example.org.',
+            id=CentralZoneTestCase.recordset__id
+        )
+
+        self.service.storage.find_recordset.return_value = recordset
+
+        # Set the zone_id value to false
+        self.service.get_recordset(
+            self.context,
+            False,
+            CentralZoneTestCase.recordset__id,
+        )
+        self.assertEqual(
+            'get_recordset',
+            designate.central.service.policy.check.call_args[0][0]
+        )
+        t, ctx, target = designate.central.service.policy.check.call_args[0]
+        self.assertEqual('get_recordset', t)
+        self.assertEqual({
+            'zone_id': CentralZoneTestCase.zone__id_2,
+            'zone_name': 'example.org.',
+            'zone_shared': self.zone_shared,
             'recordset_id': CentralZoneTestCase.recordset__id,
             'project_id': '2'}, target)
 
@@ -1172,6 +1219,7 @@ class CentralZoneTestCase(CentralBasic):
     def test_find_recordset(self):
         self.context = mock.Mock()
         self.context.project_id = 't'
+        self.service.storage.get_zone.return_value = Mockzone()
         self.service.find_recordset(self.context)
         self.assertTrue(self.service.storage.find_recordset.called)
         n, ctx, target = designate.central.service.policy.check.call_args[0]
@@ -1209,7 +1257,7 @@ class CentralZoneTestCase(CentralBasic):
 
     def test_update_recordset_action_delete(self):
         self.service.storage.get_zone.return_value = RoObject(
-            action='DELETE',
+            action='DELETE', tenant_id=''
         )
         recordset = mock.Mock(spec=objects.RecordSet)
         recordset.obj_get_changes.return_value = ['foo']
@@ -1227,6 +1275,7 @@ class CentralZoneTestCase(CentralBasic):
             name='example.org.',
             tenant_id='2',
             action='bogus',
+            shared=self.zone_shared,
         )
         recordset = mock.Mock(spec=objects.RecordSet)
         recordset.obj_get_changes.return_value = ['foo']
@@ -1247,6 +1296,7 @@ class CentralZoneTestCase(CentralBasic):
             name='example.org.',
             tenant_id='2',
             action='bogus',
+            shared=self.zone_shared,
         )
         recordset = mock.Mock(spec=objects.RecordSet)
         recordset.obj_get_changes.return_value = ['foo']
@@ -1269,7 +1319,9 @@ class CentralZoneTestCase(CentralBasic):
             'zone_id': '9c85d9b0-1e9d-4e99-aede-a06664f1af2e',
             'zone_name': 'example.org.',
             'zone_type': 'foo',
+            'zone_shared': self.zone_shared,
             'recordset_id': '9c85d9b0-1e9d-4e99-aede-a06664f1af2e',
+            'recordset_project_id': '9c85d9b0-1e9d-4e99-aede-a06664f1af2e',
             'project_id': '2'}, target)
 
     def test_update_recordset_in_storage(self):
@@ -1361,11 +1413,10 @@ class CentralZoneTestCase(CentralBasic):
             name='example.org.',
             tenant_id='2',
             type='foo',
+            shared=self.zone_shared,
         )
-        self.service.storage.get_recordset.return_value = RoObject(
+        self.service.storage.find_recordset.side_effect = (
-            zone_id=CentralZoneTestCase.zone__id,
+            exceptions.RecordSetNotFound()
-            id=CentralZoneTestCase.recordset__id,
-            managed=False,
         )
         self.context = mock.Mock()
         self.context.edit_managed_records = False
@@ -1386,7 +1437,7 @@ class CentralZoneTestCase(CentralBasic):
             tenant_id='2',
             type='foo',
         )
-        self.service.storage.get_recordset.return_value = RoObject(
+        self.service.storage.find_recordset.return_value = RoObject(
             zone_id=CentralZoneTestCase.zone__id_2,
             id=CentralZoneTestCase.recordset__id,
             managed=False,
@@ -1409,11 +1460,13 @@ class CentralZoneTestCase(CentralBasic):
             name='example.org.',
             tenant_id='2',
             type='foo',
+            shared=self.zone_shared,
         )
-        self.service.storage.get_recordset.return_value = RoObject(
+        self.service.storage.find_recordset.return_value = RoObject(
             zone_id=CentralZoneTestCase.zone__id_2,
             id=CentralZoneTestCase.recordset__id,
             managed=True,
+            tenant_id='2',
         )
         self.context = mock.Mock()
         self.context.edit_managed_records = False
@@ -1433,6 +1486,7 @@ class CentralZoneTestCase(CentralBasic):
             name='example.org.',
             tenant_id='2',
             type='foo',
+            shared=self.zone_shared,
         )
         mock_rs = objects.RecordSet(
             zone_id=CentralZoneTestCase.zone__id_2,
@@ -1442,7 +1496,7 @@ class CentralZoneTestCase(CentralBasic):
         )
 
         self.service.storage.get_zone.return_value = mock_zone
-        self.service.storage.get_recordset.return_value = mock_rs
+        self.service.storage.find_recordset.return_value = mock_rs
         self.context = mock.Mock()
         self.context.edit_managed_records = False
         self.service._delete_recordset_in_storage = mock.Mock(
@@ -1465,7 +1519,7 @@ class CentralZoneTestCase(CentralBasic):
         self.service._update_zone_in_storage = mock_uds
         self.service._delete_recordset_in_storage(
             self.context,
-            RoObject(serial=1),
+            RoObject(serial=1, shared=self.zone_shared),
             RoObject(id=2, records=[
                 RwObject(
                     action='',
@@ -1486,7 +1540,7 @@ class CentralZoneTestCase(CentralBasic):
         self.service._update_zone_in_storage = mock.Mock()
         self.service._delete_recordset_in_storage(
             self.context,
-            RoObject(serial=1),
+            RoObject(serial=1, shared=self.zone_shared),
             RoObject(id=2, records=[
                 RwObject(
                     action='',
@@ -1607,7 +1661,8 @@ class CentralZoneExportTests(CentralBasic):
 
         self.service.storage.get_zone.return_value = RoObject(
             name='example.com.',
-            id=CentralZoneTestCase.zone__id
+            id=CentralZoneTestCase.zone__id,
+            shared=False,
         )
 
         self.service.storage.create_zone_export = mock.Mock(
@@ -1756,6 +1811,7 @@ class CentralQuotaTest(unittest.TestCase):
                               'zone_records': 1,
                               'recordset_records': 1,
                               'api_export_size': 1}
+        self.zone.shared = False
 
     @patch('designate.central.service.storage')
     @patch('designate.central.service.quota')
@@ -1832,6 +1888,7 @@ class CentralQuotaTest(unittest.TestCase):
             0, 1,
             1, 1,
             1, 1,
+            1, 1,
         ]
 
         managed_recordset = mock.Mock(spec=objects.RecordSet)
@@ -1880,3 +1937,17 @@ class CentralQuotaTest(unittest.TestCase):
         # one exiting recordsets
         self.assertRaises(exceptions.OverQuota, service._enforce_record_quota,
                           self.context, self.zone, recordset_two_record)
+
+        # Test creating a recordset with a shared zone
+        mock_zone = Mockzone()
+        mock_zone.shared = True
+        service.quota.limit_check = mock.Mock()
+        service.storage.count_records = mock.Mock(return_value=1)
+
+        service._enforce_record_quota(self.context,
+                                      mock_zone,
+                                      recordset_one_record)
+
+        service.quota.limit_check.assert_called_with(self.context,
+                                                     mock_zone.tenant_id,
+                                                     recordset_records=1)

doc/source/admin/index.rst

@@ -26,4 +26,3 @@ Contents:
    troubleshooting
    samples/index
    support-matrix
-

doc/source/admin/notifications.rst

@@ -58,6 +58,8 @@ They are emitted by Central on the following events:
 * dns.zone_export.create
 * dns.zone_export.update
 * dns.zone_export.delete
+* dns.zone.share
+* dns.zone.unshare
 
 Receivers
 ---------

doc/source/user/index.rst

@@ -17,6 +17,7 @@ Managing Zones
    importexport
    zone-transfer
    secondary-zones
+   shared-zones
 
 Working with Recordsets
 -----------------------

doc/source/user/manage-zones.rst

@@ -200,3 +200,8 @@ A zone can be deleted using either its name or ID:
       +----------------+--------------------------------------+
 
 Any records present in the zone are also deleted and will no longer resolve.
+
+.. note::
+
+   Zones that have shares cannot be deleted without removing the shares or
+   using the `delete-shares` modifier.

doc/source/user/shared-zones.rst

@@ -0,0 +1,138 @@
+..
+    Copyright 2020 Cloudification GmbH.
+
+    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.
+
+Shared Zones
+============
+
+Shared zones allow sharing a particular zone across tenants. This is
+useful in cases when records for one zone should be managed by
+multiple projects. For example when a Designate zone is assigned to a
+shared network in Neutron.
+
+Zone shares have the following properties:
+
+- Quotas will be enforced against the zone owner.
+- Projects that a zone is shared with can only manage recordsets created or
+  owned by the project.
+- Zone owners can see, modify, and remove recordsets created by another
+  project.
+- Projects that a zone is shared with cannot see or modify the attributes of
+  the zone.
+- Zones that have shares cannot be deleted without removing the shares or using
+  the `delete-shares` modifier.
+- Projects that a zone is shared with cannot create sub-zones.
+
+How to Share a Zone With Another Project
+----------------------------------------
+
+Create a zone to share:
+
+.. code-block:: console
+
+    $ openstack zone create example.com. --email admin@example.com
+    +----------------+--------------------------------------+
+    | Field          | Value                                |
+    +----------------+--------------------------------------+
+    | action         | CREATE                               |
+    | email          | admin@example.com                    |
+    | id             | 92b2214f-8a57-4ed3-95f0-a64099f3b516 |
+    | name           | example.com.                         |
+    | pool_id        | 794ccc2c-d751-44fe-b57f-8894c9f5c842 |
+    | project_id     | 804806ad94364aecb0f9ae86ad653055     |
+    | serial         | 1596186919                           |
+    | status         | PENDING                              |
+    | ttl            | 3600                                 |
+    | type           | PRIMARY                              |
+    +----------------+--------------------------------------+
+
+
+Share the zone using the `openstack zone share create` command
+(in this example, the ID of the project we want to share with is
+`356df8e6c7564b5bb107f5de26cdb8ea`):
+
+.. code-block:: console
+
+    $ openstack zone share create example.com. 356df8e6c7564b5bb107f5de26cdb8ea
+    +-------------------+--------------------------------------+
+    | Field             | Value                                |
+    +-------------------+--------------------------------------+
+    | created_at        | 2023-01-30T23:17:44.000000           |
+    | id                | 77e4d5b9-2057-4be7-8cf0-9f84ef0efec1 |
+    | project_id        | 804806ad94364aecb0f9ae86ad653055     |
+    | target_project_id | 356df8e6c7564b5bb107f5de26cdb8ea     |
+    | updated_at        | None                                 |
+    | zone_id           | 92b2214f-8a57-4ed3-95f0-a64099f3b516 |
+    +-------------------+--------------------------------------+
+
+
+Project `356df8e6c7564b5bb107f5de26cdb8ea` now has access to zone
+`92b2214f-8a57-4ed3-95f0-a64099f3b516` and can manage recordsets in the zone.
+
+Using credentials for project `356df8e6c7564b5bb107f5de26cdb8ea`, we can create
+a recordset for `www.example.com.`:
+
+.. code-block:: console
+
+    $ openstack recordset create --type A --record 192.0.2.1 example.com. www
+    +-------------+--------------------------------------+
+    | Field       | Value                                |
+    +-------------+--------------------------------------+
+    | action      | CREATE                               |
+    | created_at  | 2023-01-30T23:28:05.000000           |
+    | description | None                                 |
+    | id          | aff3e00a-9e5c-4cfa-9650-65196f73418b |
+    | name        | www.example.com.                     |
+    | project_id  | 356df8e6c7564b5bb107f5de26cdb8ea     |
+    | records     | 192.0.2.1                            |
+    | status      | PENDING                              |
+    | ttl         | None                                 |
+    | type        | A                                    |
+    | updated_at  | None                                 |
+    | version     | 1                                    |
+    | zone_id     | 92b2214f-8a57-4ed3-95f0-a64099f3b516 |
+    | zone_name   | example.com.                         |
+    +-------------+--------------------------------------+
+
+
+How to List All of the Projects Sharing a Zone
+----------------------------------------------
+
+You can list all of the zone shares for a zone with the `openstack zone share
+list` command:
+
+.. code-block:: console
+
+    $ openstack zone share list example.com.
+    +-----------------------+-----------------------+-------------------------+
+    | id                    | zone_id               | target_project_id       |
+    +-----------------------+-----------------------+-------------------------+
+    | 77e4d5b9-2057-4be7-   | 92b2214f-8a57-4ed3-   | 356df8e6c7564b5bb107f5d |
+    | 8cf0-9f84ef0efec1     | 95f0-a64099f3b516     | e26cdb8ea               |
+    +-----------------------+-----------------------+-------------------------+
+
+
+How To Remove a Zone Share
+--------------------------
+
+To stop sharing a zone with a project, you can use the `openstack zone share
+delete` command:
+
+.. code-block:: console
+
+    $ openstack zone share delete example.com. 77e4d5b9-2057-4be7-8cf0-9f84ef0efec1
+
+A zone cannot be unshared in the following cases:
+
+- Zone has recordsets in other projects.

releasenotes/notes/Add-Shared-Zones-47df0368bb3ee466.yaml

@@ -0,0 +1,9 @@
+---
+features:
+  - |
+    Zones can now be shared with other projects, allowing them to create and
+    manage recordsets and records in the zone.
+other:
+  - |
+    Now that zones can be shared with multiple projects, recordsets and records
+    can have project identifiers that are different than the parent zone.