diff --git a/README.md b/README.md
index c89d721..dda601d 100644
--- a/README.md
+++ b/README.md
@@ -73,6 +73,6 @@
     
     # Wait for the VM and Mysos API endpoint to come up (http://192.168.33.17:55001 becomes available).
     
-    ./vagrant/test.sh
+    tox -e vagrant
 
 `test.sh` verifies that Mysos successfully creates a MySQL cluster and then deletes it.
diff --git a/mysos/scheduler/launcher.py b/mysos/scheduler/launcher.py
index 31f931c..beb99ca 100644
--- a/mysos/scheduler/launcher.py
+++ b/mysos/scheduler/launcher.py
@@ -8,6 +8,7 @@
 
 from .elector import MySQLMasterElector
 from .state import MySQLCluster, MySQLTask, StateProvider
+from .password import PasswordBox
 
 import mesos.interface.mesos_pb2 as mesos_pb2
 from twitter.common import log
@@ -46,6 +47,7 @@
       executor_cmd,
       election_timeout,
       admin_keypath,
+      scheduler_key,
       installer_args=None,
       backup_store_args=None,
       executor_environ=None,
@@ -61,6 +63,7 @@
       :param executor_cmd: See flags.
       :param election_timeout: See flags.
       :param admin_keypath: See flags.
+      :param scheduler_key: Used for encrypting cluster passwords.
       :param installer_args: See flags.
       :param backup_store_args: See flags.
       :param executor_environ: See flags.
@@ -97,6 +100,9 @@
     zk_root = zookeeper.parse(zk_url)[2]
     self._cluster_manager = ClusterManager(kazoo, get_cluster_path(zk_root, cluster.name))
 
+    self._password_box = PasswordBox(scheduler_key)
+    self._password_box.decrypt(cluster.encrypted_password)  # Validate the password.
+
     self._lock = threading.Lock()
 
     if self._cluster.master_id:
@@ -203,7 +209,7 @@
       cluster are killed.
     """
     with self._lock:
-      if self._cluster.password != password:
+      if not self._password_box.match(password, self._cluster.encrypted_password):
         raise self.PermissionError("No permission to kill cluster %s" % self.cluster_name)
 
       self._terminating = True
@@ -278,7 +284,7 @@
         'port': task_port,
         'cluster': self._cluster.name,
         'cluster_user': self._cluster.user,
-        'cluster_password': self._cluster.password,
+        'cluster_password': self._password_box.decrypt(self._cluster.encrypted_password),
         'server_id': server_id,  # Use the integer Task ID as the server ID.
         'zk_url': self._zk_url,
         'admin_keypath': self._admin_keypath,
diff --git a/mysos/scheduler/mysos_scheduler.py b/mysos/scheduler/mysos_scheduler.py
index 558e72d..7ff6a25 100644
--- a/mysos/scheduler/mysos_scheduler.py
+++ b/mysos/scheduler/mysos_scheduler.py
@@ -119,6 +119,12 @@
            "'local' is chosen and the state is persisted under <work_dir>/state; see --work_dir")
 
   app.add_option(
+      '--scheduler_keypath',
+      dest='scheduler_keypath',
+      help="Path to the key file that the scheduler uses to store secrets such as MySQL "
+           "cluster passwords. This key must be exactly 32 bytes long")
+
+  app.add_option(
       '--framework_failover_timeout',
       dest='framework_failover_timeout',
       default='14d',
@@ -177,6 +183,9 @@
     if not options.admin_keypath:
       app.error('Must specify --admin_keypath')
 
+    if not options.scheduler_keypath:
+      app.error('Must specify --scheduler_keypath')
+
     try:
       election_timeout = parse_time(options.election_timeout)
       framework_failover_timeout = parse_time(options.framework_failover_timeout)
@@ -206,6 +215,15 @@
       except (KeyError, yaml.YAMLError) as e:
         app.error("Invalid framework authentication key file format %s" % e)
 
+    scheduler_key = None
+    try:
+      with open(options.scheduler_keypath, 'rb') as f:
+        scheduler_key = f.read().strip()
+        if not scheduler_key:
+          raise ValueError("The key file is empty")
+    except Exception as e:
+      app.error("Cannot read --scheduler_keypath: %s" % e)
+
     log.info("Starting Mysos scheduler")
 
     kazoo = KazooClient(zk_servers)
@@ -251,6 +269,7 @@
         options.zk_url,
         election_timeout,
         options.admin_keypath,
+        scheduler_key,
         installer_args=options.installer_args,
         backup_store_args=options.backup_store_args,
         executor_environ=options.executor_environ,
diff --git a/mysos/scheduler/password.py b/mysos/scheduler/password.py
new file mode 100644
index 0000000..857a75b
--- /dev/null
+++ b/mysos/scheduler/password.py
@@ -0,0 +1,46 @@
+import random
+import string
+
+import nacl.exceptions
+import nacl.secret
+import nacl.utils
+
+
+class PasswordBox(object):
+  """
+    Implements password encryption using PyNaCl.
+  """
+
+  class Error(Exception): pass
+
+  def __init__(self, key):
+    self._secret_box = nacl.secret.SecretBox(key)
+
+  def encrypt(self, plaintext):
+    try:
+      return self._secret_box.encrypt(
+          plaintext, nacl.utils.random(nacl.secret.SecretBox.NONCE_SIZE))
+    except nacl.exceptions.CryptoError as e:
+      raise self.Error("Failed to encrypt the password: %s" % e)
+
+  def decrypt(self, encrypted):
+    try:
+      return self._secret_box.decrypt(encrypted)
+    except nacl.exceptions.CryptoError as e:
+      raise self.Error("Failed to decrypt the password: %s" % e)
+
+  def match(self, plaintext, encrypted):
+    return plaintext == self._secret_box.decrypt(encrypted)
+
+
+def gen_password():
+  """Return a randomly-generated password of 21 characters."""
+  return ''.join(random.choice(
+      string.ascii_uppercase +
+      string.ascii_lowercase +
+      string.digits) for _ in range(21))
+
+
+def gen_encryption_key():
+  """Return a randomly-generated encryption key of 32 characters."""
+  return nacl.utils.random(nacl.secret.SecretBox.KEY_SIZE)
diff --git a/mysos/scheduler/scheduler.py b/mysos/scheduler/scheduler.py
index b9eafd9..9e5d17f 100644
--- a/mysos/scheduler/scheduler.py
+++ b/mysos/scheduler/scheduler.py
@@ -2,13 +2,13 @@
 import posixpath
 import random
 import threading
-import string
 import sys
 
 from mysos.common.cluster import get_cluster_path
 from mysos.common.decorators import logged
 
 from .launcher import MySQLClusterLauncher
+from .password import gen_password, PasswordBox
 from .state import MySQLCluster, Scheduler, StateProvider
 
 import mesos.interface
@@ -44,6 +44,7 @@
       zk_url,
       election_timeout,
       admin_keypath,
+      scheduler_key,
       installer_args=None,
       backup_store_args=None,
       executor_environ=None,
@@ -58,6 +59,7 @@
       :param framework_role: See flags.
       :param election_timeout: See flags.
       :param admin_keypath: See flags.
+      :param scheduler_key: Scheduler uses it to encrypt cluster passwords.
       :param installer_args: See flags.
       :param backup_store_args: See flags.
       :param executor_environ: See flags.
@@ -91,6 +93,9 @@
     self._discover_zk_url = posixpath.join(zk_url, "discover")
     self._kazoo = kazoo
 
+    self._scheduler_key = scheduler_key
+    self._password_box = PasswordBox(scheduler_key)
+
     self._tasks = {}  # {Task ID: cluster name} mappings.
     self._launchers = OrderedDict()  # Order-preserving {cluster name : MySQLClusterLauncher}
                                      # mappings so cluster requests are fulfilled on a first come,
@@ -137,10 +142,12 @@
       self._state.clusters.add(cluster_name)
       self._state_provider.dump_scheduler_state(self._state)
 
+      # Return the plaintext version to the client but store the encrypted version.
+      password = gen_password()
       cluster = MySQLCluster(
           cluster_name,
           cluster_user,
-          gen_password(),
+          self._password_box.encrypt(password),
           int(num_nodes),
           backup_id=backup_id)
       self._state_provider.dump_cluster_state(cluster)
@@ -157,12 +164,13 @@
           self._executor_cmd,
           self._election_timeout,
           self._admin_keypath,
+          self._scheduler_key,
           installer_args=self._installer_args,
           backup_store_args=self._backup_store_args,
           executor_environ=self._executor_environ,
           framework_role=self._framework_role)
 
-      return get_cluster_path(self._discover_zk_url, cluster_name), cluster.password
+      return get_cluster_path(self._discover_zk_url, cluster_name), password
 
   def delete_cluster(self, cluster_name, password):
     """
@@ -215,7 +223,7 @@
     # connected.
     try:
       self._recover()
-    except self.Error as e:
+    except Exception as e:
       log.error("Stopping scheduler because: %s" % e)
       self._stop()
       return
@@ -230,38 +238,39 @@
     for cluster_name in OrderedSet(self._state.clusters):  # Make a copy so we can remove dead
                                                            # entries while iterating the copy.
       log.info("Recovering launcher for cluster %s" % cluster_name)
-      try:
-        cluster = self._state_provider.load_cluster_state(cluster_name)
-        if not cluster:
-          # The scheduler could have failed over before creating the launcher. The user request
-          # should have failed and there is no cluster state to restore.
-          log.info("Skipping cluster %s because its state cannot be found" % cluster_name)
-          self._state.clusters.remove(cluster_name)
-          self._state_provider.dump_scheduler_state(self._state)
-          continue
-        for task_id in cluster.tasks:
-          self._tasks[task_id] = cluster.name  # Reconstruct the 'tasks' map.
-        # Order of launchers is preserved thanks to the OrderedSet.
-        # For recovered launchers we use the currently specified --framework_role and
-        # --executor_environ, etc., instead of saving it in cluster state so the change in flags can
-        # be picked up by existing clusters.
-        self._launchers[cluster.name] = MySQLClusterLauncher(
-            self._driver,
-            cluster,
-            self._state_provider,
-            self._discover_zk_url,
-            self._kazoo,
-            self._framework_user,
-            self._executor_uri,
-            self._executor_cmd,
-            self._election_timeout,
-            self._admin_keypath,
-            self._installer_args,
-            self._backup_store_args,
-            self._executor_environ,
-            self._framework_role)
-      except StateProvider.Error as e:
-        raise self.Error("Failed to recover cluster: %s" % e.message)
+
+      cluster = self._state_provider.load_cluster_state(cluster_name)
+      if not cluster:
+        # The scheduler could have failed over before creating the launcher. The user request
+        # should have failed and there is no cluster state to restore.
+        log.info("Skipping cluster %s because its state cannot be found" % cluster_name)
+        self._state.clusters.remove(cluster_name)
+        self._state_provider.dump_scheduler_state(self._state)
+        continue
+
+      for task_id in cluster.tasks:
+        self._tasks[task_id] = cluster.name  # Reconstruct the 'tasks' map.
+
+      # Order of launchers is preserved thanks to the OrderedSet.
+      # For recovered launchers we use the currently specified --framework_role and
+      # --executor_environ, etc., instead of saving it in cluster state so the change in flags can
+      # be picked up by existing clusters.
+      self._launchers[cluster.name] = MySQLClusterLauncher(
+          self._driver,
+          cluster,
+          self._state_provider,
+          self._discover_zk_url,
+          self._kazoo,
+          self._framework_user,
+          self._executor_uri,
+          self._executor_cmd,
+          self._election_timeout,
+          self._admin_keypath,
+          self._scheduler_key,
+          self._installer_args,
+          self._backup_store_args,
+          self._executor_environ,
+          self._framework_role)
 
     log.info("Recovered %s clusters" % len(self._launchers))
 
@@ -371,11 +380,3 @@
   copy = li[:]
   random.shuffle(copy)
   return copy
-
-
-def gen_password():
-  """Return a randomly-generated password of 21 characters."""
-  return ''.join(random.choice(
-      string.ascii_uppercase +
-      string.ascii_lowercase +
-      string.digits) for _ in range(21))
diff --git a/mysos/scheduler/state.py b/mysos/scheduler/state.py
index 5e8efd6..62763fc 100644
--- a/mysos/scheduler/state.py
+++ b/mysos/scheduler/state.py
@@ -85,13 +85,13 @@
     It includes tasks (MySQLTask) for members of the cluster.
   """
 
-  def __init__(self, name, user, password, num_nodes, backup_id=None):
+  def __init__(self, name, user, encrypted_password, num_nodes, backup_id=None):
     if not isinstance(num_nodes, int):
       raise TypeError("'num_nodes' should be an int")
 
     self.name = name
     self.user = user
-    self.password = password
+    self.encrypted_password = encrypted_password
     self.num_nodes = num_nodes
     self.backup_id = backup_id
 
@@ -187,7 +187,6 @@
     except PickleError as e:
       raise self.Error('Failed to recover MySQLCluster: %s' % e)
 
-  @abstractmethod
   def remove_cluster_state(self, cluster_name):
     path = self._get_cluster_state_path(cluster_name)
     if not os.path.isfile(path):
diff --git a/setup.py b/setup.py
index 59b10f6..ab01bf7 100644
--- a/setup.py
+++ b/setup.py
@@ -41,14 +41,10 @@
              list_package_data_files('mysos/scheduler', 'assets'))
     },
     install_requires=[
-        'cherrypy==3.2.2',
         'kazoo==1.3.1',
         'mako==0.4.0',
         'mesos.interface{0}'.format(MESOS_VERSION),
-        'mysql-python',
         'pyyaml==3.10',
-        'sqlalchemy',
-        'zake==0.2.1',
         make_commons_requirement('app'),
         make_commons_requirement('collections'),
         make_commons_requirement('concurrent'),
@@ -60,15 +56,30 @@
         make_commons_requirement('zookeeper'),
     ],
     extras_require={
-        'test': ['webtest',],
-        'driver': ['mesos.native{0}'.format(MESOS_VERSION),],
+        'test': [
+            'pynacl>=0.3.0',
+            'webtest',
+            'zake==0.2.1',
+        ],
+        'scheduler': [
+            'cherrypy==3.2.2',
+            'mesos.native{0}'.format(MESOS_VERSION),
+            'pynacl>=0.3.0,<1',
+        ],
+        'executor': [
+            'mesos.native{0}'.format(MESOS_VERSION),
+        ],
+        'test_client': [
+            'sqlalchemy',
+            'mysql-python'
+        ]
     },
     entry_points={
         'console_scripts': [
-            'mysos_scheduler=mysos.scheduler.mysos_scheduler:proxy_main [driver]',
-            'mysos_executor=mysos.executor.mysos_executor:proxy_main [driver]',
-            'vagrant_mysos_executor=mysos.executor.testing.vagrant_mysos_executor:proxy_main [driver]',
-            'mysos_test_client=mysos.testing.mysos_test_client:proxy_main',
+            'mysos_scheduler=mysos.scheduler.mysos_scheduler:proxy_main [scheduler]',
+            'mysos_executor=mysos.executor.mysos_executor:proxy_main [executor]',
+            'vagrant_mysos_executor=mysos.executor.testing.vagrant_mysos_executor:proxy_main [executor]',
+            'mysos_test_client=mysos.testing.mysos_test_client:proxy_main [test_client]',
         ],
     },
 )
diff --git a/tests/scheduler/test_launcher.py b/tests/scheduler/test_launcher.py
index 9dd4bd6..e6ea5d4 100644
--- a/tests/scheduler/test_launcher.py
+++ b/tests/scheduler/test_launcher.py
@@ -7,6 +7,7 @@
 from mysos.common.cluster import get_cluster_path, wait_for_master
 from mysos.common.testing import Fake
 from mysos.scheduler.launcher import create_resources, MySQLClusterLauncher
+from mysos.scheduler.password import gen_encryption_key, PasswordBox
 from mysos.scheduler.state import LocalStateProvider, MySQLCluster
 from mysos.scheduler.zk_state import ZooKeeperStateProvider
 
@@ -53,7 +54,11 @@
 
     # Some tests use the default launcher; some don't.
     self._zk_url = "zk://host/mysos/test"
-    self._cluster = MySQLCluster("cluster0", "user", "pass", 3)
+
+    self._scheduler_key = gen_encryption_key()
+    self._password_box = PasswordBox(self._scheduler_key)
+
+    self._cluster = MySQLCluster("cluster0", "user", self._password_box.encrypt("pass"), 3)
 
     # Construct the state provider based on the test parameter.
     if request.param == LocalStateProvider:
@@ -74,6 +79,7 @@
         "cmd.sh",
         Amount(5, Time.SECONDS),
         "/etc/mysos/admin_keyfile.yml",
+        self._scheduler_key,
         query_interval=Amount(150, Time.MILLISECONDS))  # Short interval.
 
     self._elected = threading.Event()
@@ -166,7 +172,7 @@
     launchers = [
       MySQLClusterLauncher(
           self._driver,
-          MySQLCluster("cluster0", "user0", "pass0", 1),
+          MySQLCluster("cluster0", "user0", self._password_box.encrypt("pass0"), 1),
           self._state_provider,
           self._zk_url,
           self._zk_client,
@@ -174,10 +180,11 @@
           "./executor.pex",
           "cmd.sh",
           Amount(5, Time.SECONDS),
-          "/etc/mysos/admin_keyfile.yml"),
+          "/etc/mysos/admin_keyfile.yml",
+          self._scheduler_key),
       MySQLClusterLauncher(
           self._driver,
-          MySQLCluster("cluster1", "user1", "pass1", 2),
+          MySQLCluster("cluster1", "user1", self._password_box.encrypt("pass1"), 2),
           self._state_provider,
           self._zk_url,
           self._zk_client,
@@ -185,7 +192,8 @@
           "./executor.pex",
           "cmd.sh",
           Amount(5, Time.SECONDS),
-          "/etc/mysos/admin_keyfile.yml")]
+          "/etc/mysos/admin_keyfile.yml",
+          self._scheduler_key)]
     self._launchers.extend(launchers)
 
     resources = create_resources(cpus=4, mem=512 * 3, ports=set([10000, 10001, 10002]))
@@ -218,7 +226,8 @@
         "./executor.pex",
         "cmd.sh",
         Amount(5, Time.SECONDS),
-        "/etc/mysos/admin_keyfile.yml")
+        "/etc/mysos/admin_keyfile.yml",
+        self._scheduler_key)
     self._launchers.append(launcher)
 
     resources = create_resources(cpus=4, mem=512 * 3, ports=set([10000]))
@@ -253,7 +262,8 @@
         "./executor.pex",
         "cmd.sh",
         Amount(1, Time.SECONDS),
-        "/etc/mysos/admin_keyfile.yml")
+        "/etc/mysos/admin_keyfile.yml",
+        self._scheduler_key)
     self._launchers.append(launcher)
 
     resources = create_resources(cpus=4, mem=512 * 3, ports=set([10000]))
@@ -405,6 +415,7 @@
         "cmd.sh",
         Amount(5, Time.SECONDS),
         "/etc/mysos/admin_keyfile.yml",
+        self._scheduler_key,
         query_interval=Amount(150, Time.MILLISECONDS))
 
     # Now fail the master task.
@@ -480,6 +491,7 @@
         "cmd.sh",
         Amount(5, Time.SECONDS),
         "/etc/mysos/admin_keyfile.yml",
+        self._scheduler_key,
         query_interval=Amount(150, Time.MILLISECONDS))
 
     for i in range(1, self._cluster.num_nodes):
@@ -535,7 +547,8 @@
     with pytest.raises(MySQLClusterLauncher.PermissionError):
       self._launcher.kill("wrong_password")
 
-    self._launcher.kill(self._cluster.password)  # Correct password.
+    # Correct password.
+    self._launcher.kill(self._password_box.decrypt(self._cluster.encrypted_password))
 
     # All 3 nodes are successfully killed.
     status = mesos_pb2.TaskStatus()
@@ -547,3 +560,37 @@
 
     assert "/mysos/test/cluster0" not in self._storage.paths  # ServerSets removed.
     assert not self._state_provider.load_cluster_state("cluster0")  # State removed.
+
+  def test_launcher_recovery_corrupted_password(self):
+    # 1. Launch a single instance for a cluster on the running launcher.
+    task_id, remaining = self._launcher.launch(self._offer)
+    del self._offer.resources[:]
+    self._offer.resources.extend(remaining)
+    assert task_id == "mysos-cluster0-0"
+
+    # The task has successfully started.
+    status = mesos_pb2.TaskStatus()
+    status.state = mesos_pb2.TASK_RUNNING
+    status.slave_id.value = self._offer.slave_id.value
+    status.task_id.value = "mysos-cluster0-0"
+    self._launcher.status_update(status)
+
+    # 2. Recover the launcher.
+    self._cluster = self._state_provider.load_cluster_state(self._cluster.name)
+    self._cluster.encrypted_password = "corrupted_password"
+
+    # The corrupted password causes the launcher constructor to fail.
+    with pytest.raises(ValueError):
+      self._launcher = MySQLClusterLauncher(
+          self._driver,
+          self._cluster,
+          self._state_provider,
+          self._zk_url,
+          self._zk_client,
+          self._framework_user,
+          "./executor.pex",
+          "cmd.sh",
+          Amount(5, Time.SECONDS),
+          "/etc/mysos/admin_keyfile.yml",
+          self._scheduler_key,
+          query_interval=Amount(150, Time.MILLISECONDS))
diff --git a/tests/scheduler/test_mysos_scheduler.py b/tests/scheduler/test_mysos_scheduler.py
index 6338371..a4796e8 100644
--- a/tests/scheduler/test_mysos_scheduler.py
+++ b/tests/scheduler/test_mysos_scheduler.py
@@ -3,6 +3,7 @@
 import posixpath
 
 from mysos.common.cluster import get_cluster_path, wait_for_master
+from mysos.scheduler.password import gen_encryption_key
 from mysos.scheduler.scheduler import MysosScheduler
 from mysos.scheduler.state import LocalStateProvider, Scheduler
 
@@ -60,7 +61,8 @@
       zk_client,
       zk_url,
       Amount(40, Time.SECONDS),
-      "/fakepath")
+      "/fakepath",
+      gen_encryption_key())
 
   scheduler_driver = mesos.native.MesosSchedulerDriver(
       scheduler,
diff --git a/tests/scheduler/test_scheduler.py b/tests/scheduler/test_scheduler.py
index 4b967c8..7ef9910 100644
--- a/tests/scheduler/test_scheduler.py
+++ b/tests/scheduler/test_scheduler.py
@@ -9,6 +9,7 @@
     INCOMPATIBLE_ROLE_OFFER_REFUSE_DURATION,
     MysosScheduler)
 from mysos.scheduler.launcher import create_resources
+from mysos.scheduler.password import gen_encryption_key, PasswordBox
 from mysos.scheduler.state import LocalStateProvider, MySQLCluster, Scheduler
 
 from kazoo.handlers.threading import SequentialThreadingHandler
@@ -68,6 +69,8 @@
     shutil.rmtree(self._tmpdir, True)  # Clean up after ourselves.
 
   def test_scheduler_recovery(self):
+    scheduler_key = gen_encryption_key()
+
     scheduler1 = MysosScheduler(
         self._state,
         self._state_provider,
@@ -77,7 +80,8 @@
         self._zk_client,
         self._zk_url,
         Amount(5, Time.SECONDS),
-        "/etc/mysos/admin_keyfile.yml")
+        "/etc/mysos/admin_keyfile.yml",
+        scheduler_key)
     scheduler1.registered(self._driver, self._framework_id, object())
     scheduler1.create_cluster("cluster1", "mysql_user", 3)
     scheduler1.resourceOffers(self._driver, [self._offer])
@@ -102,7 +106,8 @@
         self._zk_client,
         self._zk_url,
         Amount(5, Time.SECONDS),
-        "/etc/mysos/admin_keyfile.yml")
+        "/etc/mysos/admin_keyfile.yml",
+        scheduler_key)
 
     # Scheduler always receives registered() with the same FrameworkID after failover.
     scheduler2.registered(self._driver, self._framework_id, object())
@@ -115,6 +120,8 @@
       scheduler2.create_cluster("cluster1", "mysql_user", 3)
 
   def test_scheduler_recovery_failure_before_launch(self):
+    scheduler_key = gen_encryption_key()
+
     scheduler1 = MysosScheduler(
         self._state,
         self._state_provider,
@@ -124,9 +131,10 @@
         self._zk_client,
         self._zk_url,
         Amount(5, Time.SECONDS),
-        "/etc/mysos/admin_keyfile.yml")
+        "/etc/mysos/admin_keyfile.yml",
+        scheduler_key)
     scheduler1.registered(self._driver, self._framework_id, object())
-    scheduler1.create_cluster("cluster1", "mysql_user", 3)
+    _, password = scheduler1.create_cluster("cluster1", "mysql_user", 3)
 
     # Simulate restart before the task is successfully launched.
     scheduler2 = MysosScheduler(
@@ -138,7 +146,8 @@
         self._zk_client,
         self._zk_url,
         Amount(5, Time.SECONDS),
-        "/etc/mysos/admin_keyfile.yml")
+        "/etc/mysos/admin_keyfile.yml",
+        scheduler_key)
 
     assert len(scheduler2._launchers) == 0  # No launchers are recovered.
 
@@ -148,6 +157,11 @@
     assert len(scheduler2._launchers) == 1
     assert scheduler2._launchers["cluster1"].cluster_name == "cluster1"
 
+    password_box = PasswordBox(scheduler_key)
+
+    assert password_box.match(
+        password, scheduler2._launchers["cluster1"]._cluster.encrypted_password)
+
     # Now offer the resources for this task.
     scheduler2.resourceOffers(self._driver, [self._offer])
 
@@ -169,6 +183,7 @@
         self._zk_url,
         Amount(5, Time.SECONDS),
         "/etc/mysos/admin_keyfile.yml",
+        gen_encryption_key(),
         framework_role='mysos')  # Require 'mysos' but the resources are in '*'.
     scheduler1.registered(self._driver, self._framework_id, object())
     scheduler1.create_cluster("cluster1", "mysql_user", 3)
diff --git a/tests/scheduler/test_state.py b/tests/scheduler/test_state.py
index 692d974..43bed59 100644
--- a/tests/scheduler/test_state.py
+++ b/tests/scheduler/test_state.py
@@ -9,6 +9,7 @@
     MySQLTask,
     Scheduler
 )
+from mysos.scheduler.password import gen_encryption_key, PasswordBox
 
 from mesos.interface.mesos_pb2 import FrameworkInfo
 
@@ -44,7 +45,9 @@
     assert expected.clusters == actual.clusters
 
   def test_cluster_state(self):
-    expected = MySQLCluster('cluster1', 'cluster_user', 'cluster_password', 3)
+    password_box = PasswordBox(gen_encryption_key())
+
+    expected = MySQLCluster('cluster1', 'cluster_user', password_box.encrypt('cluster_password'), 3)
 
     expected.tasks['task1'] = MySQLTask(
         'cluster1', 'task1', 'slave1', 'host1', 10000)
@@ -57,3 +60,5 @@
     assert expected.num_nodes == actual.num_nodes
     assert len(expected.tasks) == len(actual.tasks)
     assert expected.tasks['task1'].port == actual.tasks['task1'].port
+    assert expected.encrypted_password == actual.encrypted_password
+    assert password_box.match('cluster_password', actual.encrypted_password)
diff --git a/tox.ini b/tox.ini
index cf6f944..7d4fd2d 100644
--- a/tox.ini
+++ b/tox.ini
@@ -14,7 +14,7 @@
 whitelist_externals=mkdir
 commands =
   mkdir -p {toxinidir}/dist/
-  pip install --find-links {toxinidir}/3rdparty -e .[driver]
+  pip install --find-links {toxinidir}/3rdparty -e .[scheduler]
   pex \
     --source-dir={toxinidir} \
     --output-file={toxinidir}/dist/fake_mysos_executor.pex \
@@ -32,3 +32,9 @@
   twitter.checkstyle==0.1.0
 skip_install = True
 commands = twitterstyle -n ImportOrder mysos tests
+
+# This currently requires the Vagrant VM to be up.
+# TODO(jyx): Launch Vagrant here directly.
+[testenv:vagrant]
+install_command = pip install -e .[test_client] --find-links {toxinidir}/3rdparty {opts} {packages}
+commands = {toxinidir}/vagrant/test.sh
diff --git a/vagrant/bin/mysos_executor.sh b/vagrant/bin/mysos_executor.sh
index d9738f1..474cd0e 100755
--- a/vagrant/bin/mysos_executor.sh
+++ b/vagrant/bin/mysos_executor.sh
@@ -6,7 +6,11 @@
 
 # Using python to run pip and vagrant_mysos_executor because the shebang in venv/bin/pip can
 # exceed system limit and cannot be executed directly.
+
+# 'protobuf' is a a dependency of mesos.interface's but we install it separately because otherwise
+# 3.0.0-alpha is installed and it breaks the mesos.interface install.
+venv/bin/python venv/bin/pip install 'protobuf==2.6.1'
 venv/bin/python venv/bin/pip install --find-links /home/vagrant/mysos/deps mesos.native
-venv/bin/python venv/bin/pip install --pre --find-links . mysos
+venv/bin/python venv/bin/pip install --pre --find-links . mysos[executor]
 
 venv/bin/python venv/bin/vagrant_mysos_executor
diff --git a/vagrant/bin/mysos_scheduler.sh b/vagrant/bin/mysos_scheduler.sh
index e8148af..c050591 100755
--- a/vagrant/bin/mysos_scheduler.sh
+++ b/vagrant/bin/mysos_scheduler.sh
@@ -5,8 +5,12 @@
 TMPDIR=$(mktemp -d)
 
 virtualenv $TMPDIR  # Create venv under /tmp.
+
+# 'protobuf' is a dependency of mesos.interface's but we install it separately because otherwise
+# 3.0.0-alpha is installed and it breaks the mesos.interface install.
+$TMPDIR/bin/pip install 'protobuf==2.6.1'
 $TMPDIR/bin/pip install --find-links /home/vagrant/mysos/deps mesos.native
-$TMPDIR/bin/pip install --pre --find-links /home/vagrant/mysos/dist mysos
+$TMPDIR/bin/pip install --pre --find-links /home/vagrant/mysos/dist mysos[scheduler]
 
 ZK_HOST=192.168.33.17
 API_PORT=55001
@@ -24,4 +28,5 @@
     --framework_failover_timeout=1m \
     --framework_role=mysos \
     --framework_authentication_file=/home/vagrant/mysos/vagrant/etc/fw_auth_keyfile.yml \
+    --scheduler_keypath=/home/vagrant/mysos/vagrant/etc/scheduler_keyfile.txt \
     --executor_environ='[{"name": "MYSOS_DEFAULTS_FILE", "value": "/etc/mysql/conf.d/my5.6.cnf"}]'
diff --git a/vagrant/etc/scheduler_keyfile.txt b/vagrant/etc/scheduler_keyfile.txt
new file mode 100644
index 0000000..0e4a2ed
--- /dev/null
+++ b/vagrant/etc/scheduler_keyfile.txt
@@ -0,0 +1 @@
+73SZAptK4K6i2sB8fw6B0aQf0qLO6zmw
\ No newline at end of file
diff --git a/vagrant/provision-dev-cluster.sh b/vagrant/provision-dev-cluster.sh
index d8390cf..55a0b17 100755
--- a/vagrant/provision-dev-cluster.sh
+++ b/vagrant/provision-dev-cluster.sh
@@ -6,19 +6,15 @@
 export DEBIAN_FRONTEND=noninteractive
 aptitude update -q
 aptitude install -q -y -o Dpkg::Options::="--force-confdef" -o Dpkg::Options::="--force-confold" \
-    libcurl3-dev \
-    libsasl2-dev \
+    curl \  # We use curl --silent to download pakcages.
+    libcurl3-dev \  # Mesos requirement.
+    libsasl2-dev \  # Mesos requirement.
     python-dev \
     zookeeper \
     mysql-server-5.6 \
     libmysqlclient-dev \
-    libunwind8 \
     python-virtualenv \
-    bison flex  # For libnl.
-
-# Fix up a dependency issue of Mesos egg: _mesos.so links to libunwind.so.7 but Trusty only has
-# libunwind.so.8.
-ln -sf /usr/lib/x86_64-linux-gnu/libunwind.so.8 /usr/lib/x86_64-linux-gnu/libunwind.so.7
+    libffi-dev  # For pynacl.
 
 # Fix up Ubuntu mysql-server-5.6 issue: mysql_install_db looks for this file even if we don't need
 # it.
@@ -38,7 +34,6 @@
 
 # Install the upstart configurations.
 sudo cp /home/vagrant/mysos/vagrant/upstart/*.conf /etc/init
-chown -R vagrant:vagrant /home/vagrant/mysos
 EOF
 chmod +x /usr/local/bin/update-mysos
 sudo -u vagrant update-mysos
diff --git a/vagrant/test.sh b/vagrant/test.sh
index a0b69c9..bbfe21f 100755
--- a/vagrant/test.sh
+++ b/vagrant/test.sh
@@ -11,7 +11,7 @@
 cluster_user="mysos"
 
 HERE="$(cd "$(dirname "$0")" && pwd)"
-executable=$HERE/../.tox/py27/bin/mysos_test_client
+executable=$HERE/../.tox/vagrant/bin/mysos_test_client
 
 if [ ! -f ${executable} ]; then
   echo "${executable} doesn't exist. Build it first."