diff --git a/designate/api/v1/extensions/quotas.py b/designate/api/v1/extensions/quotas.py
new file mode 100644
index 000000000..4860a4ca2
--- /dev/null
+++ b/designate/api/v1/extensions/quotas.py
@@ -0,0 +1,52 @@
+# Copyright 2012 Hewlett-Packard Development Company, L.P.
+#
+# Author: Kiall Mac Innes <kiall@hp.com>
+#
+# 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.
+import flask
+from designate.openstack.common import log as logging
+from designate.central import rpcapi as central_rpcapi
+
+LOG = logging.getLogger(__name__)
+central_api = central_rpcapi.CentralAPI()
+blueprint = flask.Blueprint('quotas', __name__)
+
+
+@blueprint.route('/quotas/<tenant_id>', methods=['GET'])
+def get_quotas(tenant_id):
+    context = flask.request.environ.get('context')
+
+    quotas = central_api.get_quotas(context, tenant_id)
+
+    return flask.jsonify(quotas)
+
+
+@blueprint.route('/quotas/<tenant_id>', methods=['PUT', 'POST'])
+def set_quota(tenant_id):
+    context = flask.request.environ.get('context')
+    values = flask.request.json
+
+    for resource, hard_limit in values.items():
+        central_api.set_quota(context, tenant_id, resource, hard_limit)
+
+    quotas = central_api.get_quotas(context, tenant_id)
+    return flask.jsonify(quotas)
+
+
+@blueprint.route('/quotas/<tenant_id>', methods=['DELETE'])
+def reset_quotas(tenant_id):
+    context = flask.request.environ.get('context')
+
+    central_api.reset_quotas(context, tenant_id)
+
+    return flask.Response(status=200)
diff --git a/designate/central/__init__.py b/designate/central/__init__.py
index 88bd1883a..054d00ecd 100644
--- a/designate/central/__init__.py
+++ b/designate/central/__init__.py
@@ -65,8 +65,6 @@ cfg.CONF.register_opts([
                help='The backend driver to use'),
     cfg.StrOpt('storage-driver', default='sqlalchemy',
                help='The storage driver to use'),
-    cfg.StrOpt('quota-driver', default='storage',
-               help='The quota driver to use'),
     cfg.ListOpt('enabled-notification-handlers', default=[],
                 help='Enabled Notification Handlers'),
     cfg.ListOpt('domain-name-blacklist',
diff --git a/designate/central/rpcapi.py b/designate/central/rpcapi.py
index 68b12e49b..4cc0a74dc 100644
--- a/designate/central/rpcapi.py
+++ b/designate/central/rpcapi.py
@@ -31,6 +31,7 @@ class CentralAPI(rpc_proxy.RpcProxy):
         1.2 - Add get_tenant and get_tenants
         1.3 - Add get_absolute_limits
         2.0 - Renamed most get_resources to find_resources
+        2.1 - Add quota methods
 
     """
     def __init__(self, topic=None):
@@ -43,6 +44,29 @@ class CentralAPI(rpc_proxy.RpcProxy):
 
         return self.call(context, msg)
 
+    # Quota Methods
+    def get_quotas(self, context, tenant_id):
+        msg = self.make_msg('get_quotas', tenant_id=tenant_id)
+
+        return self.call(context, msg, version='2.1')
+
+    def get_quota(self, context, tenant_id, resource):
+        msg = self.make_msg('get_quota', tenant_id=tenant_id,
+                            resource=resource)
+
+        return self.call(context, msg, version='2.1')
+
+    def set_quota(self, context, tenant_id, resource, hard_limit):
+        msg = self.make_msg('set_quota', tenant_id=tenant_id,
+                            resource=resource, hard_limit=hard_limit)
+
+        return self.call(context, msg, version='2.1')
+
+    def reset_quotas(self, context, tenant_id):
+        msg = self.make_msg('reset_quotas', tenant_id=tenant_id)
+
+        return self.call(context, msg, version='2.1')
+
     # Server Methods
     def create_server(self, context, values):
         msg = self.make_msg('create_server', values=values)
diff --git a/designate/central/service.py b/designate/central/service.py
index 5885bed85..cfb1a4bd6 100644
--- a/designate/central/service.py
+++ b/designate/central/service.py
@@ -43,7 +43,7 @@ def wrap_backend_call():
 
 
 class Service(rpc_service.Service):
-    RPC_API_VERSION = '2.0'
+    RPC_API_VERSION = '2.1'
 
     def __init__(self, *args, **kwargs):
         backend_driver = cfg.CONF['service:central'].backend_driver
@@ -254,7 +254,38 @@ class Service(rpc_service.Service):
 
     # Misc Methods
     def get_absolute_limits(self, context):
-        return self.quota.get_tenant_quotas(context, context.tenant_id)
+        # NOTE(Kiall): Currently, we only have quota based limits..
+        return self.quota.get_quotas(context, context.tenant_id)
+
+    # Quota Methods
+    def get_quotas(self, context, tenant_id):
+        target = {'tenant_id': tenant_id}
+        policy.check('get_quotas', context, target)
+
+        return self.quota.get_quotas(context, tenant_id)
+
+    def get_quota(self, context, tenant_id, resource):
+        target = {'tenant_id': tenant_id, 'resource': resource}
+        policy.check('get_quota', context, target)
+
+        return self.quota.get_quota(context, tenant_id, resource)
+
+    def set_quota(self, context, tenant_id, resource, hard_limit):
+        target = {
+            'tenant_id': tenant_id,
+            'resource': resource,
+            'hard_limit': hard_limit,
+        }
+
+        policy.check('set_quota', context, target)
+
+        return self.quota.set_quota(context, tenant_id, resource, hard_limit)
+
+    def reset_quotas(self, context, tenant_id):
+        target = {'tenant_id': tenant_id}
+        policy.check('reset_quotas', context, target)
+
+        self.quota.reset_quotas(context, tenant_id)
 
     # Server Methods
     def create_server(self, context, values):
diff --git a/designate/quota/__init__.py b/designate/quota/__init__.py
index 6280ed3c6..00247a7c0 100644
--- a/designate/quota/__init__.py
+++ b/designate/quota/__init__.py
@@ -20,6 +20,7 @@ from designate.quota.base import Quota
 LOG = logging.getLogger(__name__)
 
 cfg.CONF.register_opts([
+    cfg.StrOpt('quota-driver', default='storage', help='Quota driver to use'),
     cfg.IntOpt('quota-domains', default=10, help='Number of domains allowed '
                                                  'per tenant'),
     cfg.IntOpt('quota-domain-records', default=500, help='Number of records '
@@ -28,6 +29,4 @@ cfg.CONF.register_opts([
 
 
 def get_quota():
-    quota_driver = cfg.CONF['service:central'].quota_driver
-
-    return Quota.get_plugin(quota_driver, invoke_on_load=True)
+    return Quota.get_plugin(cfg.CONF.quota_driver, invoke_on_load=True)
diff --git a/designate/quota/base.py b/designate/quota/base.py
index de6d84fd8..3f4b79388 100644
--- a/designate/quota/base.py
+++ b/designate/quota/base.py
@@ -25,24 +25,8 @@ class Quota(Plugin):
     __plugin_ns__ = 'designate.quota'
     __plugin_type__ = 'quota'
 
-    def get_tenant_quotas(self, context, tenant_id):
-        """
-        Get quotas for a tenant.
-
-        :param context: RPC Context.
-        :param tenant_id: Tenant ID to fetch quotas for
-        """
-        quotas = {
-            'domains': cfg.CONF.quota_domains,
-            'domain_records': cfg.CONF.quota_domain_records,
-        }
-
-        quotas.update(self._get_tenant_quotas(context, tenant_id))
-
-        return quotas
-
     def limit_check(self, context, tenant_id, **values):
-        quotas = self.get_tenant_quotas(context, tenant_id)
+        quotas = self.get_quotas(context, tenant_id)
 
         for resource, value in values.items():
             if resource in quotas:
@@ -51,6 +35,33 @@ class Quota(Plugin):
             else:
                 raise exceptions.QuotaResourceUnknown()
 
+    def get_quotas(self, context, tenant_id):
+        quotas = self.get_default_quotas(context)
+
+        quotas.update(self._get_quotas(context, tenant_id))
+
+        return quotas
+
     @abc.abstractmethod
-    def _get_tenant_quotas(self, context, tenant_id):
+    def _get_quotas(self, context, tenant_id):
         pass
+
+    def get_default_quotas(self, context):
+        return {
+            'domains': cfg.CONF.quota_domains,
+            'domain_records': cfg.CONF.quota_domain_records,
+        }
+
+    def get_quota(self, context, tenant_id, resource):
+        quotas = self._get_quotas(context, tenant_id)
+
+        if resource not in quotas:
+            raise exceptions.QuotaResourceUnknown()
+
+        return quotas[resource]
+
+    def set_quota(self, context, tenant_id, resource, hard_limit):
+        raise NotImplementedError()
+
+    def reset_quotas(self, context, tenant_id):
+        raise NotImplementedError()
diff --git a/designate/quota/impl_config.py b/designate/quota/impl_noop.py
similarity index 72%
rename from designate/quota/impl_config.py
rename to designate/quota/impl_noop.py
index ff339b6d2..f537274bb 100644
--- a/designate/quota/impl_config.py
+++ b/designate/quota/impl_noop.py
@@ -19,9 +19,8 @@ from designate.quota.base import Quota
 LOG = logging.getLogger(__name__)
 
 
-class ConfigQuota(Quota):
-    def _get_tenant_quotas(self, context, tenant_id):
-        # NOTE(kiall): The base class handles merging the config defaults into
-        #              the per-tenent values we return. Since we only care
-        #              about the config values, we return an empty dict.
+class NoopQuota(Quota):
+    __plugin_name__ = 'noop'
+
+    def _get_quotas(self, context, tenant_id):
         return {}
diff --git a/designate/quota/impl_storage.py b/designate/quota/impl_storage.py
index 7260cc1e8..e7c14b469 100644
--- a/designate/quota/impl_storage.py
+++ b/designate/quota/impl_storage.py
@@ -13,25 +13,78 @@
 # 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 import exceptions
 from designate.openstack.common import log as logging
 from designate.quota.base import Quota
-from designate import storage
+from designate.storage import api as sapi
 
 LOG = logging.getLogger(__name__)
 
 
 class StorageQuota(Quota):
-    def __init__(self):
+    __plugin_name__ = 'storage'
+
+    def __init__(self, storage_api=None):
         super(StorageQuota, self).__init__()
 
-        # Get a storage connection
-        self.storage = storage.get_storage()
+        if storage_api is None:
+            storage_api = sapi.StorageAPI()
 
-    def _get_tenant_quotas(self, context, tenant_id):
-        criterion = dict(
-            tenant_id=tenant_id
-        )
+        self.storage_api = storage_api
 
-        quotas = self.storage.find_quotas(context, criterion)
+    def _get_quotas(self, context, tenant_id):
+        quotas = self.storage_api.find_quotas(context, {
+            'tenant_id': tenant_id,
+        })
 
         return dict((q['resource'], q['hard_limit']) for q in quotas)
+
+    def get_quota(self, context, tenant_id, resource):
+        quota = self.storage_api.find_quota(context, {
+            'tenant_id': tenant_id,
+            'resource': resource,
+        })
+
+        return {resource: quota['hard_limit']}
+
+    def set_quota(self, context, tenant_id, resource, hard_limit):
+        def create_quota():
+            values = {
+                'tenant_id': tenant_id,
+                'resource': resource,
+                'hard_limit': hard_limit,
+            }
+
+            with self.storage_api.create_quota(context, values):
+                pass  # NOTE(kiall): No other systems need updating.
+
+        def update_quota():
+            values = {'hard_limit': hard_limit}
+
+            with self.storage_api.update_quota(context, quota['id'], values):
+                pass  # NOTE(kiall): No other systems need updating.
+
+        if resource not in self.get_default_quotas(context).keys():
+            raise exceptions.QuotaResourceUnknown("%s is not a valid quota "
+                                                  "resource", resource)
+
+        try:
+            quota = self.storage_api.find_quota(context, {
+                'tenant_id': tenant_id,
+                'resource': resource,
+            })
+        except exceptions.NotFound:
+            create_quota()
+        else:
+            update_quota()
+
+        return {resource: hard_limit}
+
+    def reset_quotas(self, context, tenant_id):
+        quotas = self.storage_api.find_quotas(context, {
+            'tenant_id': tenant_id,
+        })
+
+        for quota in quotas:
+            with self.storage_api.delete_quota(context, quota['id']):
+                pass  # NOTE(kiall): No other systems need updating.
diff --git a/designate/storage/api.py b/designate/storage/api.py
index 4b41f42a6..037dc487e 100644
--- a/designate/storage/api.py
+++ b/designate/storage/api.py
@@ -51,7 +51,7 @@ class StorageAPI(object):
         """
         return self.storage.get_quota(context, quota_id)
 
-    def find_quotas(self, context, criterion):
+    def find_quotas(self, context, criterion=None):
         """
         Find Quotas
 
diff --git a/designate/tests/test_quota/__init__.py b/designate/tests/test_quota/__init__.py
index b7a9a17cd..136abd3ef 100644
--- a/designate/tests/test_quota/__init__.py
+++ b/designate/tests/test_quota/__init__.py
@@ -29,10 +29,10 @@ class QuotaTestCase(TestCase):
         super(QuotaTestCase, self).setUp()
         self.quota = quota.get_quota()
 
-    def test_get_tenant_quotas(self):
+    def test_get_quotas(self):
         context = self.get_admin_context()
 
-        quotas = self.quota.get_tenant_quotas(context, 'DefaultQuotaTenant')
+        quotas = self.quota.get_quotas(context, 'DefaultQuotaTenant')
 
         self.assertIsNotNone(quotas)
         self.assertEqual(quotas, {
diff --git a/designate/tests/test_quota/test_config.py b/designate/tests/test_quota/test_noop.py
similarity index 84%
rename from designate/tests/test_quota/test_config.py
rename to designate/tests/test_quota/test_noop.py
index 65e73d72b..6923c6b60 100644
--- a/designate/tests/test_quota/test_config.py
+++ b/designate/tests/test_quota/test_noop.py
@@ -19,9 +19,9 @@ from designate.tests.test_quota import QuotaTestCase
 LOG = logging.getLogger(__name__)
 
 
-class ConfigQuotaTest(QuotaTestCase):
+class NoopQuotaTest(QuotaTestCase):
     __test__ = True
 
     def setUp(self):
-        self.config(quota_driver='config', group='service:central')
-        super(ConfigQuotaTest, self).setUp()
+        self.config(quota_driver='noop')
+        super(NoopQuotaTest, self).setUp()
diff --git a/designate/tests/test_quota/test_storage.py b/designate/tests/test_quota/test_storage.py
index 0bea25307..6ae904b33 100644
--- a/designate/tests/test_quota/test_storage.py
+++ b/designate/tests/test_quota/test_storage.py
@@ -23,5 +23,68 @@ class StorageQuotaTest(QuotaTestCase):
     __test__ = True
 
     def setUp(self):
-        self.config(quota_driver='storage', group='service:central')
+        self.config(quota_driver='storage')
         super(StorageQuotaTest, self).setUp()
+
+    def test_set_quota_create(self):
+        quota = self.quota.set_quota(self.admin_context, 'tenant_id',
+                                     'domains', 1500)
+
+        self.assertEquals(quota, {'domains': 1500})
+
+        # Drop into the storage layer directly to ensure the quota was created
+        # sucessfully.
+        criterion = {
+            'tenant_id': 'tenant_id',
+            'resource': 'domains'
+        }
+
+        quota = self.quota.storage_api.find_quota(self.admin_context,
+                                                  criterion)
+
+        self.assertEquals(quota['tenant_id'], 'tenant_id')
+        self.assertEquals(quota['resource'], 'domains')
+        self.assertEquals(quota['hard_limit'], 1500)
+
+    def test_set_quota_update(self):
+        # First up, Create the quota
+        self.quota.set_quota(self.admin_context, 'tenant_id', 'domains', 1500)
+
+        # Next, update the quota
+        self.quota.set_quota(self.admin_context, 'tenant_id', 'domains', 1234)
+
+        # Drop into the storage layer directly to ensure the quota was updated
+        # sucessfully
+        criterion = {
+            'tenant_id': 'tenant_id',
+            'resource': 'domains'
+        }
+
+        quota = self.quota.storage_api.find_quota(self.admin_context,
+                                                  criterion)
+
+        self.assertEquals(quota['tenant_id'], 'tenant_id')
+        self.assertEquals(quota['resource'], 'domains')
+        self.assertEquals(quota['hard_limit'], 1234)
+
+    def test_reset_quotas(self):
+        # First up, Create a domains quota
+        self.quota.set_quota(self.admin_context, 'tenant_id', 'domains', 1500)
+
+        # Then, Create a domain_records quota
+        self.quota.set_quota(self.admin_context, 'tenant_id', 'domain_records',
+                             800)
+
+        # Now, Reset the tenants quota
+        self.quota.reset_quotas(self.admin_context, 'tenant_id')
+
+        # Drop into the storage layer directly to ensure the tenant has no
+        # specific quotas registed.
+        criterion = {
+            'tenant_id': 'tenant_id'
+        }
+
+        quotas = self.quota.storage_api.find_quotas(self.admin_context,
+                                                    criterion)
+
+        self.assertEquals(0, len(quotas))
diff --git a/etc/designate/designate.conf.sample b/etc/designate/designate.conf.sample
index c9d216c56..51e3b12d9 100644
--- a/etc/designate/designate.conf.sample
+++ b/etc/designate/designate.conf.sample
@@ -62,7 +62,7 @@ default_log_levels = amqplib=WARN, sqlalchemy=WARN, boto=WARN, suds=INFO, keysto
 #auth_strategy = noauth
 
 # Enabled API Version 1 extensions
-#enabled_extensions_v1 = diagnostics, sync, import, export, reports
+#enabled_extensions_v1 = diagnostics, quotas, reports, sync
 
 #-----------------------
 # Agent Service
diff --git a/etc/designate/policy.json b/etc/designate/policy.json
index be7f09320..572fa5bfc 100644
--- a/etc/designate/policy.json
+++ b/etc/designate/policy.json
@@ -5,6 +5,11 @@
 
     "default": "rule:admin_or_owner",
 
+    "get_quotas": "rule:admin_or_owner",
+    "get_quota": "rule:admin_or_owner",
+    "set_quota": "rule:admin",
+    "reset_quotas": "rule:admin",
+
     "create_server": "rule:admin",
     "find_servers": "rule:admin",
     "get_server": "rule:admin",
diff --git a/setup.cfg b/setup.cfg
index 79331a796..7f8dd87c0 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -52,6 +52,7 @@ designate.api.v1 =
 
 designate.api.v1.extensions =
     diagnostics = designate.api.v1.extensions.diagnostics:blueprint
+    quotas = designate.api.v1.extensions.quotas:blueprint
     sync = designate.api.v1.extensions.sync:blueprint
     reports = designate.api.v1.extensions.reports:blueprint
 
@@ -71,8 +72,8 @@ designate.backend =
     fake = designate.backend.impl_fake:FakeBackend
 
 designate.quota =
+    noop =  designate.quota.impl_noop:NoopQuota
     storage = designate.quota.impl_storage:StorageQuota
-    config =  designate.quota.impl_config:ConfigQuota
 
 designate.manage =
     database-init = designate.manage.database:InitCommand