diff --git a/etc/nova/policy.json b/etc/nova/policy.json
index 81b7b4d293d3..afc0ef6b819e 100644
--- a/etc/nova/policy.json
+++ b/etc/nova/policy.json
@@ -11,6 +11,7 @@
     "compute:create:forced_host": "is_admin:True",
     "compute:get_all": "",
     "compute:get_all_tenants": "",
+    "compute:unlock_override": "rule:admin_api",
 
     "compute:shelve": "",
     "compute:shelve_offload": "",
@@ -23,8 +24,8 @@
     "compute_extension:admin_actions:unpause": "rule:admin_or_owner",
     "compute_extension:admin_actions:suspend": "rule:admin_or_owner",
     "compute_extension:admin_actions:resume": "rule:admin_or_owner",
-    "compute_extension:admin_actions:lock": "rule:admin_api",
-    "compute_extension:admin_actions:unlock": "rule:admin_api",
+    "compute_extension:admin_actions:lock": "rule:admin_or_owner",
+    "compute_extension:admin_actions:unlock": "rule:admin_or_owner",
     "compute_extension:admin_actions:resetNetwork": "rule:admin_api",
     "compute_extension:admin_actions:injectNetworkInfo": "rule:admin_api",
     "compute_extension:admin_actions:createBackup": "rule:admin_or_owner",
@@ -36,8 +37,8 @@
     "compute_extension:v3:os-admin-actions:unpause": "rule:admin_or_owner",
     "compute_extension:v3:os-admin-actions:suspend": "rule:admin_or_owner",
     "compute_extension:v3:os-admin-actions:resume": "rule:admin_or_owner",
-    "compute_extension:v3:os-admin-actions:lock": "rule:admin_api",
-    "compute_extension:v3:os-admin-actions:unlock": "rule:admin_api",
+    "compute_extension:v3:os-admin-actions:lock": "rule:admin_or_owner",
+    "compute_extension:v3:os-admin-actions:unlock": "rule:admin_or_owner",
     "compute_extension:v3:os-admin-actions:reset_network": "rule:admin_api",
     "compute_extension:v3:os-admin-actions:inject_network_info": "rule:admin_api",
     "compute_extension:v3:os-admin-actions:create_backup": "rule:admin_or_owner",
diff --git a/nova/api/openstack/compute/contrib/admin_actions.py b/nova/api/openstack/compute/contrib/admin_actions.py
index 3554eede82cc..f4fdb00168b7 100644
--- a/nova/api/openstack/compute/contrib/admin_actions.py
+++ b/nova/api/openstack/compute/contrib/admin_actions.py
@@ -161,7 +161,7 @@ class AdminActionsController(wsgi.Controller):
 
     @wsgi.action('lock')
     def _lock(self, req, id, body):
-        """Permit admins to lock a server."""
+        """Lock a server instance."""
         context = req.environ['nova.context']
         authorize(context, 'lock')
         try:
@@ -177,12 +177,14 @@ class AdminActionsController(wsgi.Controller):
 
     @wsgi.action('unlock')
     def _unlock(self, req, id, body):
-        """Permit admins to unlock a server."""
+        """Unlock a server instance."""
         context = req.environ['nova.context']
         authorize(context, 'unlock')
         try:
             instance = self.compute_api.get(context, id)
             self.compute_api.unlock(context, instance)
+        except exception.PolicyNotAuthorized as e:
+            raise webob.exc.HTTPForbidden(explanation=e.format_message())
         except exception.InstanceNotFound:
             raise exc.HTTPNotFound(_("Server not found"))
         except Exception:
diff --git a/nova/api/openstack/compute/plugins/v3/admin_actions.py b/nova/api/openstack/compute/plugins/v3/admin_actions.py
index 45548fa72e60..ec551a0ad4a3 100644
--- a/nova/api/openstack/compute/plugins/v3/admin_actions.py
+++ b/nova/api/openstack/compute/plugins/v3/admin_actions.py
@@ -172,7 +172,7 @@ class AdminActionsController(wsgi.Controller):
     @extensions.expected_errors(404)
     @wsgi.action('lock')
     def _lock(self, req, id, body):
-        """Permit admins to lock a server."""
+        """Lock a server instance."""
         context = req.environ['nova.context']
         authorize(context, 'lock')
         try:
@@ -185,7 +185,7 @@ class AdminActionsController(wsgi.Controller):
     @extensions.expected_errors(404)
     @wsgi.action('unlock')
     def _unlock(self, req, id, body):
-        """Permit admins to lock a server."""
+        """Unlock a server instance."""
         context = req.environ['nova.context']
         authorize(context, 'unlock')
         try:
diff --git a/nova/api/openstack/compute/plugins/v3/extended_status.py b/nova/api/openstack/compute/plugins/v3/extended_status.py
index 94c87fadb2ea..854767a2ce00 100644
--- a/nova/api/openstack/compute/plugins/v3/extended_status.py
+++ b/nova/api/openstack/compute/plugins/v3/extended_status.py
@@ -29,7 +29,7 @@ class ExtendedStatusController(wsgi.Controller):
         self.compute_api = compute.API()
 
     def _extend_server(self, server, instance):
-        for state in ['task_state', 'vm_state', 'power_state']:
+        for state in ['task_state', 'vm_state', 'power_state', 'locked_by']:
             key = "%s:%s" % (ExtendedStatus.alias, state)
             server[key] = instance[state]
 
@@ -84,6 +84,8 @@ def make_server(elem):
              '%s:power_state' % ExtendedStatus.alias)
     elem.set('{%s}vm_state' % ExtendedStatus.namespace,
              '%s:vm_state' % ExtendedStatus.alias)
+    elem.set('{%s}locked_by' % ExtendedStatus.namespace,
+             '%s:locked_by' % ExtendedStatus.alias)
 
 
 class ExtendedStatusTemplate(xmlutil.TemplateBuilder):
diff --git a/nova/compute/api.py b/nova/compute/api.py
index 55ab190beb8b..93868a9de970 100644
--- a/nova/compute/api.py
+++ b/nova/compute/api.py
@@ -2473,18 +2473,33 @@ class API(base.Base):
     @wrap_check_policy
     def lock(self, context, instance):
         """Lock the given instance."""
+        # Only update the lock if we are an admin (non-owner)
+        is_owner = instance['project_id'] == context.project_id
+        if instance['locked'] and is_owner:
+            return
+
         context = context.elevated()
         instance_uuid = instance['uuid']
         LOG.debug(_('Locking'), context=context, instance_uuid=instance_uuid)
-        self._instance_update(context, instance_uuid, locked=True)
+        self._instance_update(context, instance_uuid, locked=True,
+                              locked_by='owner' if is_owner else 'admin')
 
     @wrap_check_policy
     def unlock(self, context, instance):
         """Unlock the given instance."""
+        # If the instance was locked by someone else, check
+        # that we're allowed to override the lock
+        is_owner = instance['project_id'] == context.project_id
+        expect_locked_by = 'owner' if is_owner else 'admin'
+        locked_by = instance['locked_by']
+        if locked_by and locked_by != expect_locked_by:
+            check_policy(context, 'unlock_override', instance)
+
         context = context.elevated()
         instance_uuid = instance['uuid']
         LOG.debug(_('Unlocking'), context=context, instance_uuid=instance_uuid)
-        self._instance_update(context, instance_uuid, locked=False)
+        self._instance_update(context, instance_uuid, locked=False,
+                              locked_by=None)
 
     @wrap_check_policy
     def get_lock(self, context, instance):
diff --git a/nova/db/sqlalchemy/migrate_repo/versions/205_add_locked_by_to_instance.py b/nova/db/sqlalchemy/migrate_repo/versions/205_add_locked_by_to_instance.py
new file mode 100644
index 000000000000..efc4f35f0aaf
--- /dev/null
+++ b/nova/db/sqlalchemy/migrate_repo/versions/205_add_locked_by_to_instance.py
@@ -0,0 +1,46 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 OpenStack Foundation.
+#
+#    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 sqlalchemy import Column, MetaData, Enum, Table
+from sqlalchemy.dialects import postgresql
+
+
+def upgrade(migrate_engine):
+    meta = MetaData()
+    meta.bind = migrate_engine
+    for table_name in ['instances', 'shadow_instances']:
+        enum = Enum('owner', 'admin', name='%s0locked_by' % table_name)
+        if migrate_engine.url.get_dialect() is postgresql.dialect:
+            # Need to explicitly create Postgres enums during migrations
+            enum.create(migrate_engine, checkfirst=False)
+
+        instances = Table(table_name, meta, autoload=True)
+        locked_by = Column('locked_by', enum)
+        instances.create_column(locked_by)
+        instances.update().\
+            where(instances.c.locked == True).\
+            values(locked_by='admin').execute()
+
+
+def downgrade(migrate_engine):
+    meta = MetaData()
+    meta.bind = migrate_engine
+    for table_name in ['instances', 'shadow_instances']:
+        instances = Table(table_name, meta, autoload=True)
+        instances.drop_column('locked_by')
+
+        Enum(name='%s0locked_by' % table_name).drop(migrate_engine,
+                                                    checkfirst=False)
diff --git a/nova/db/sqlalchemy/models.py b/nova/db/sqlalchemy/models.py
index c33c391798a5..ee8c4bfcc5e2 100644
--- a/nova/db/sqlalchemy/models.py
+++ b/nova/db/sqlalchemy/models.py
@@ -21,7 +21,7 @@
 SQLAlchemy models for nova data.
 """
 
-from sqlalchemy import Column, Index, Integer, BigInteger, String, schema
+from sqlalchemy import Column, Index, Integer, BigInteger, Enum, String, schema
 from sqlalchemy.dialects.mysql import MEDIUMTEXT
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy import ForeignKey, DateTime, Boolean, Text, Float
@@ -246,7 +246,11 @@ class Instance(BASE, NovaBase):
     # To remember on which host an instance booted.
     # An instance may have moved to another host by live migration.
     launched_on = Column(Text, nullable=True)
+
+    # NOTE(jdillaman): locked deprecated in favor of locked_by,
+    # to be removed in Icehouse
     locked = Column(Boolean, nullable=True)
+    locked_by = Column(Enum('owner', 'admin'), nullable=True)
 
     os_type = Column(String(255), nullable=True)
     architecture = Column(String(255), nullable=True)
diff --git a/nova/objects/instance.py b/nova/objects/instance.py
index 323cec2b9404..0c25aa294dbc 100644
--- a/nova/objects/instance.py
+++ b/nova/objects/instance.py
@@ -47,7 +47,8 @@ class Instance(base.NovaObject):
     # Version 1.2: Added security_groups
     # Version 1.3: Added expected_vm_state and admin_state_reset to
     #              save()
-    VERSION = '1.3'
+    # Version 1.4: Added locked_by and deprecated locked
+    VERSION = '1.4'
 
     fields = {
         'id': int,
@@ -92,7 +93,11 @@ class Instance(base.NovaObject):
         'display_description': obj_utils.str_or_none,
 
         'launched_on': obj_utils.str_or_none,
+
+        # NOTE(jdillaman): locked deprecated in favor of locked_by,
+        # to be removed in Icehouse
         'locked': bool,
+        'locked_by': obj_utils.str_or_none,
 
         'os_type': obj_utils.str_or_none,
         'architecture': obj_utils.str_or_none,
diff --git a/nova/tests/api/openstack/compute/contrib/test_admin_actions.py b/nova/tests/api/openstack/compute/contrib/test_admin_actions.py
index 1d9e1094803c..a66f21e5124b 100644
--- a/nova/tests/api/openstack/compute/contrib/test_admin_actions.py
+++ b/nova/tests/api/openstack/compute/contrib/test_admin_actions.py
@@ -329,6 +329,21 @@ class AdminActionsTest(test.TestCase):
             unicode(exception.DestinationHypervisorTooOld()),
             res.body)
 
+    def test_unlock_not_authorized(self):
+        def fake_unlock(*args, **kwargs):
+            raise exception.PolicyNotAuthorized(action='unlock')
+
+        self.stubs.Set(compute_api.API, 'unlock', fake_unlock)
+
+        app = fakes.wsgi_app(init_only=('servers',))
+        req = webob.Request.blank('/v2/fake/servers/%s/action' %
+                self.UUID)
+        req.method = 'POST'
+        req.body = jsonutils.dumps({'unlock': None})
+        req.content_type = 'application/json'
+        res = req.get_response(app)
+        self.assertEqual(res.status_int, 403)
+
 
 class CreateBackupTests(test.TestCase):
 
diff --git a/nova/tests/api/openstack/compute/plugins/v3/test_admin_actions.py b/nova/tests/api/openstack/compute/plugins/v3/test_admin_actions.py
index 6a03967bcb78..23a735dac58b 100644
--- a/nova/tests/api/openstack/compute/plugins/v3/test_admin_actions.py
+++ b/nova/tests/api/openstack/compute/plugins/v3/test_admin_actions.py
@@ -312,6 +312,21 @@ class AdminActionsTest(test.TestCase):
                                    'disk_over_commit': False}})
         self.assertEqual(res.status_int, 409)
 
+    def test_unlock_not_authorized(self):
+        def fake_unlock(*args, **kwargs):
+            raise exception.PolicyNotAuthorized(action='unlock')
+
+        self.stubs.Set(compute_api.API, 'unlock', fake_unlock)
+
+        app = fakes.wsgi_app_v3(init_only=('servers', 'os-admin-actions'))
+        req = webob.Request.blank('/v3/servers/%s/action' %
+                self.UUID)
+        req.method = 'POST'
+        req.body = jsonutils.dumps({'unlock': None})
+        req.content_type = 'application/json'
+        res = req.get_response(app)
+        self.assertEqual(res.status_int, 403)
+
 
 class CreateBackupTests(test.TestCase):
 
diff --git a/nova/tests/api/openstack/compute/plugins/v3/test_extended_status.py b/nova/tests/api/openstack/compute/plugins/v3/test_extended_status.py
index 722254309d6d..6270ed050411 100644
--- a/nova/tests/api/openstack/compute/plugins/v3/test_extended_status.py
+++ b/nova/tests/api/openstack/compute/plugins/v3/test_extended_status.py
@@ -32,15 +32,15 @@ UUID3 = '00000000-0000-0000-0000-000000000003'
 
 def fake_compute_get(*args, **kwargs):
     return fakes.stub_instance(1, uuid=UUID3, task_state="kayaking",
-            vm_state="slightly crunchy", power_state=1)
+            vm_state="slightly crunchy", power_state=1, locked_by='owner')
 
 
 def fake_compute_get_all(*args, **kwargs):
     db_list = [
         fakes.stub_instance(1, uuid=UUID1, task_state="task-1",
-                vm_state="vm-1", power_state=1),
+                vm_state="vm-1", power_state=1, locked_by=None),
         fakes.stub_instance(2, uuid=UUID2, task_state="task-2",
-                vm_state="vm-2", power_state=2),
+                vm_state="vm-2", power_state=2, locked_by='admin'),
     ]
     fields = instance_obj.INSTANCE_DEFAULT_FIELDS
     return instance_obj._make_instance_list(args[1],
@@ -73,11 +73,14 @@ class ExtendedStatusTest(test.TestCase):
     def _get_servers(self, body):
         return jsonutils.loads(body).get('servers')
 
-    def assertServerStates(self, server, vm_state, power_state, task_state):
+    def assertServerStates(self, server, vm_state, power_state, task_state,
+                           locked_by):
         self.assertEqual(server.get('%svm_state' % self.prefix), vm_state)
         self.assertEqual(int(server.get('%spower_state' % self.prefix)),
                          power_state)
         self.assertEqual(server.get('%stask_state' % self.prefix), task_state)
+        self.assertEqual(str(server.get('%slocked_by' % self.prefix)),
+                         locked_by)
 
     def test_show(self):
         url = '/v3/servers/%s' % UUID3
@@ -87,7 +90,8 @@ class ExtendedStatusTest(test.TestCase):
         self.assertServerStates(self._get_server(res.body),
                                 vm_state='slightly crunchy',
                                 power_state=1,
-                                task_state='kayaking')
+                                task_state='kayaking',
+                                locked_by='owner')
 
     def test_detail(self):
         url = '/v3/servers/detail'
@@ -98,7 +102,8 @@ class ExtendedStatusTest(test.TestCase):
             self.assertServerStates(server,
                                     vm_state='vm-%s' % (i + 1),
                                     power_state=(i + 1),
-                                    task_state='task-%s' % (i + 1))
+                                    task_state='task-%s' % (i + 1),
+                                    locked_by=['None', 'admin'][i])
 
     def test_no_instance_passthrough_404(self):
 
diff --git a/nova/tests/api/openstack/fakes.py b/nova/tests/api/openstack/fakes.py
index f4bf932be9b5..5602e83debd9 100644
--- a/nova/tests/api/openstack/fakes.py
+++ b/nova/tests/api/openstack/fakes.py
@@ -471,7 +471,7 @@ def stub_instance(id, user_id=None, project_id=None, host=None,
                   limit=None, marker=None,
                   launched_at=timeutils.utcnow(),
                   terminated_at=timeutils.utcnow(),
-                  availability_zone=''):
+                  availability_zone='', locked_by=None):
 
     if user_id is None:
         user_id = 'fake_user'
@@ -545,7 +545,8 @@ def stub_instance(id, user_id=None, project_id=None, host=None,
         "availability_zone": availability_zone,
         "display_name": display_name or server_name,
         "display_description": "",
-        "locked": False,
+        "locked": locked_by != None,
+        "locked_by": locked_by,
         "metadata": metadata,
         "access_ip_v4": access_ipv4,
         "access_ip_v6": access_ipv6,
diff --git a/nova/tests/compute/test_compute.py b/nova/tests/compute/test_compute.py
index 81945dedf952..fd8a512b2d7a 100644
--- a/nova/tests/compute/test_compute.py
+++ b/nova/tests/compute/test_compute.py
@@ -2693,6 +2693,48 @@ class ComputeTestCase(BaseTestCase):
         self.compute.terminate_instance(self.context,
                 instance=jsonutils.to_primitive(instance))
 
+    def _check_locked_by(self, instance_uuid, locked_by):
+        instance = db.instance_get_by_uuid(self.context, instance_uuid)
+        self.assertEqual(instance['locked'], locked_by != None)
+        self.assertEqual(instance['locked_by'], locked_by)
+        return instance
+
+    def test_override_owner_lock(self):
+        admin_context = context.RequestContext('admin-user',
+                                               'admin-project',
+                                               is_admin=True)
+
+        instance = jsonutils.to_primitive(self._create_fake_instance())
+        instance_uuid = instance['uuid']
+        self.compute.run_instance(self.context, instance=instance)
+
+        # Ensure that an admin can override the owner lock
+        self.compute_api.lock(self.context, instance)
+        self._check_locked_by(instance_uuid, 'owner')
+        self.compute_api.unlock(admin_context, instance)
+        self._check_locked_by(instance_uuid, None)
+
+    def test_upgrade_owner_lock(self):
+        admin_context = context.RequestContext('admin-user',
+                                               'admin-project',
+                                               is_admin=True)
+
+        instance = jsonutils.to_primitive(self._create_fake_instance())
+        instance_uuid = instance['uuid']
+        self.compute.run_instance(self.context, instance=instance)
+
+        # Ensure that an admin can upgrade the lock and that
+        # the owner can no longer unlock
+        self.compute_api.lock(self.context, instance)
+        self.compute_api.lock(admin_context, instance)
+        instance = self._check_locked_by(instance_uuid, 'admin')
+        self.assertRaises(exception.PolicyNotAuthorized,
+                          self.compute_api.unlock,
+                          self.context, instance)
+        self._check_locked_by(instance_uuid, 'admin')
+        self.compute_api.unlock(admin_context, instance)
+        self._check_locked_by(instance_uuid, None)
+
     def _test_state_revert(self, instance, operation, pre_task_state,
                            post_task_state=None, kwargs=None):
         if kwargs is None:
diff --git a/nova/tests/db/test_migrations.py b/nova/tests/db/test_migrations.py
index a44b5be89a11..96d2b3c80343 100644
--- a/nova/tests/db/test_migrations.py
+++ b/nova/tests/db/test_migrations.py
@@ -2433,6 +2433,28 @@ class TestNovaMigrations(BaseMigrationTestCase, CommonTestsMixIn):
         else:
             self.assertIn(('reservations_uuid_idx', ['uuid']), index_data)
 
+    def _pre_upgrade_205(self, engine):
+        fake_instances = [dict(uuid='m205-uuid1', locked=True),
+                          dict(uuid='m205-uuid2', locked=False)]
+        for table_name in ['instances', 'shadow_instances']:
+            table = db_utils.get_table(engine, table_name)
+            engine.execute(table.insert(), fake_instances)
+
+    def _check_205(self, engine, data):
+        for table_name in ['instances', 'shadow_instances']:
+            table = db_utils.get_table(engine, table_name)
+            rows = table.select().\
+                where(table.c.uuid.in_(['m205-uuid1', 'm205-uuid2'])).\
+                order_by(table.c.uuid).execute().fetchall()
+            self.assertEqual(rows[0]['locked_by'], 'admin')
+            self.assertEqual(rows[1]['locked_by'], None)
+
+    def _post_downgrade_205(self, engine):
+        for table_name in ['instances', 'shadow_instances']:
+            table = db_utils.get_table(engine, table_name)
+            rows = table.select().execute().fetchall()
+            self.assertFalse('locked_by' in rows[0])
+
 
 class TestBaremetalMigrations(BaseMigrationTestCase, CommonTestsMixIn):
     """Test sqlalchemy-migrate migrations."""
diff --git a/nova/tests/fake_policy.py b/nova/tests/fake_policy.py
index 177bc9a5b7a3..2afe4330bda9 100644
--- a/nova/tests/fake_policy.py
+++ b/nova/tests/fake_policy.py
@@ -44,6 +44,7 @@ policy_data = """
     "compute:get_lock": "",
     "compute:lock": "",
     "compute:unlock": "",
+    "compute:unlock_override": "is_admin:True",
 
     "compute:get_vnc_console": "",
     "compute:get_spice_console": "",