diff --git a/conf/st2.dev.conf b/conf/st2.dev.conf
index cf2b5b6596..bcda4aeb07 100644
--- a/conf/st2.dev.conf
+++ b/conf/st2.dev.conf
@@ -40,6 +40,7 @@ logging = st2reactor/conf/logging.timersengine.conf
 [actionrunner]
 logging = st2actions/conf/logging.conf
 stream_output = True
+encryption_key_path = conf/st2_kvstore_demo.crypto.key.json
 
 [rbac]
 enable = False
diff --git a/st2api/tests/unit/controllers/v1/test_action_alias.py b/st2api/tests/unit/controllers/v1/test_action_alias.py
index bf31469c57..b11f7b0169 100644
--- a/st2api/tests/unit/controllers/v1/test_action_alias.py
+++ b/st2api/tests/unit/controllers/v1/test_action_alias.py
@@ -13,6 +13,18 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from oslo_config import cfg
+
+opts = [
+    cfg.StrOpt(
+        "encryption_key_path",
+        default="conf/st2_kvstore_demo.crypto.key.json",
+        help="Location of the symmetric encryption key for encrypting values in kvstore. "
+        "This key should be in JSON and should've been generated using "
+        "st2-generate-symmetric-crypto-key tool.",
+    ),
+]
+cfg.CONF.register_opts(opts, group="actionrunner")
 from st2common.models.api.action import ActionAliasAPI
 from st2api.controllers.v1.actionalias import ActionAliasController
 
diff --git a/st2api/tests/unit/controllers/v1/test_action_views.py b/st2api/tests/unit/controllers/v1/test_action_views.py
index 2f6be710bf..1b33609f8f 100644
--- a/st2api/tests/unit/controllers/v1/test_action_views.py
+++ b/st2api/tests/unit/controllers/v1/test_action_views.py
@@ -21,6 +21,19 @@
 import mock
 
 from st2common.content import utils as content_utils
+from oslo_config import cfg
+
+opts = [
+    cfg.StrOpt(
+        "encryption_key_path",
+        default="conf/st2_kvstore_demo.crypto.key.json",
+        help="Location of the symmetric encryption key for encrypting values in kvstore. "
+        "This key should be in JSON and should've been generated using "
+        "st2-generate-symmetric-crypto-key tool.",
+    ),
+]
+cfg.CONF.register_opts(opts, group="actionrunner")
+
 import st2common.validators.api.action as action_validator
 from st2common.util.compat import mock_open_name
 from st2api.controllers.v1.action_views import OverviewController
diff --git a/st2api/tests/unit/controllers/v1/test_actions.py b/st2api/tests/unit/controllers/v1/test_actions.py
index c1891125d7..f4e3b3104f 100644
--- a/st2api/tests/unit/controllers/v1/test_actions.py
+++ b/st2api/tests/unit/controllers/v1/test_actions.py
@@ -28,6 +28,18 @@
 import mock
 import unittest2
 from six.moves import http_client
+from oslo_config import cfg
+
+opts = [
+    cfg.StrOpt(
+        "encryption_key_path",
+        default="conf/st2_kvstore_demo.crypto.key.json",
+        help="Location of the symmetric encryption key for encrypting values in kvstore. "
+        "This key should be in JSON and should've been generated using "
+        "st2-generate-symmetric-crypto-key tool.",
+    ),
+]
+cfg.CONF.register_opts(opts, group="actionrunner")
 
 from st2common.persistence.action import Action
 import st2common.validators.api.action as action_validator
diff --git a/st2api/tests/unit/controllers/v1/test_alias_execution.py b/st2api/tests/unit/controllers/v1/test_alias_execution.py
index 44261fde3f..d8256eed47 100644
--- a/st2api/tests/unit/controllers/v1/test_alias_execution.py
+++ b/st2api/tests/unit/controllers/v1/test_alias_execution.py
@@ -18,6 +18,18 @@
 import mock
 
 from st2common.constants.action import LIVEACTION_STATUS_SUCCEEDED
+from oslo_config import cfg
+
+opts = [
+    cfg.StrOpt(
+        "encryption_key_path",
+        default="conf/st2_kvstore_demo.crypto.key.json",
+        help="Location of the symmetric encryption key for encrypting values in kvstore. "
+        "This key should be in JSON and should've been generated using "
+        "st2-generate-symmetric-crypto-key tool.",
+    ),
+]
+cfg.CONF.register_opts(opts, group="actionrunner")
 from st2common.models.db.execution import ActionExecutionDB
 from st2common.services import action as action_service
 from st2tests.api import SUPER_SECRET_PARAMETER
diff --git a/st2api/tests/unit/controllers/v1/test_executions.py b/st2api/tests/unit/controllers/v1/test_executions.py
index eb3face2ba..2bc0fc78cf 100644
--- a/st2api/tests/unit/controllers/v1/test_executions.py
+++ b/st2api/tests/unit/controllers/v1/test_executions.py
@@ -30,6 +30,18 @@
 from st2common.content import utils as content_utils
 from st2common.models.api.keyvalue import KeyValuePairAPI
 from st2common.models.db.auth import UserDB
+from oslo_config import cfg
+
+opts = [
+    cfg.StrOpt(
+        "encryption_key_path",
+        default="conf/st2_kvstore_demo.crypto.key.json",
+        help="Location of the symmetric encryption key for encrypting values in kvstore. "
+        "This key should be in JSON and should've been generated using "
+        "st2-generate-symmetric-crypto-key tool.",
+    ),
+]
+cfg.CONF.register_opts(opts, group="actionrunner")
 from st2common.models.db.execution import ActionExecutionDB
 from st2common.models.db.execution import ActionExecutionOutputDB
 from st2common.models.db.keyvalue import KeyValuePairDB
@@ -1728,7 +1740,6 @@ def test_get_include_attributes_and_secret_parameters(self):
             ("/v1/actionexecutions?include_attributes=parameters,runner"),
             ("/v1/actionexecutions?include_attributes=parameters,action,runner"),
         ]
-
         for url in urls:
             resp = self.app.get(url + "&limit=1&show_secrets=True")
 
diff --git a/st2common/st2common/config.py b/st2common/st2common/config.py
index c88955e4bb..a795a78f06 100644
--- a/st2common/st2common/config.py
+++ b/st2common/st2common/config.py
@@ -494,6 +494,13 @@ def register_opts(ignore_errors=False):
                 "that size"
             ),
         ),
+        cfg.StrOpt(
+            "encryption_key_path",
+            default="conf/st2_kvstore_demo.crypto.key.json",
+            help="Location of the symmetric encryption key for encrypting values in kvstore. "
+            "This key should be in JSON and should've been generated using "
+            "st2-generate-symmetric-crypto-key tool.",
+        ),
     ]
 
     do_register_opts(
diff --git a/st2common/st2common/models/db/execution.py b/st2common/st2common/models/db/execution.py
index 0de35a5c31..182e3f2e9c 100644
--- a/st2common/st2common/models/db/execution.py
+++ b/st2common/st2common/models/db/execution.py
@@ -19,23 +19,24 @@
 
 import mongoengine as me
 
-from st2common import log as logging
 from st2common.models.db import stormbase
 from st2common.fields import JSONDictEscapedFieldCompatibilityField
 from st2common.fields import ComplexDateTimeField
 from st2common.util import date as date_utils
 from st2common.util import output_schema
+from oslo_config import cfg
+
 from st2common.util.secrets import get_secret_parameters
 from st2common.util.secrets import mask_inquiry_response
 from st2common.util.secrets import mask_secret_parameters
+from st2common.util.secrets import encrypt_secret_parameters
+from st2common.util.crypto import read_crypto_key
+
 from st2common.constants.types import ResourceType
 
 __all__ = ["ActionExecutionDB", "ActionExecutionOutputDB"]
 
 
-LOG = logging.getLogger(__name__)
-
-
 class ActionExecutionDB(stormbase.StormFoundationDB):
     RESOURCE_TYPE = ResourceType.EXECUTION
     UID_FIELDS = ["id"]
@@ -99,6 +100,7 @@ class ActionExecutionDB(stormbase.StormFoundationDB):
             {"fields": ["task_execution"]},
         ]
     }
+    encryption_key = read_crypto_key(cfg.CONF.actionrunner.encryption_key_path)
 
     def get_uid(self):
         # TODO Construct id from non id field:
@@ -178,6 +180,89 @@ def get_masked_parameters(self):
         serializable_dict = self.to_serializable_dict(mask_secrets=True)
         return serializable_dict["parameters"]
 
+    def save(self, *args, **kwargs):
+        original_parameters = copy.deepcopy(self.parameters)
+        parameters = {}
+        parameters.update(dict(self.action).get("parameters", {}))
+        parameters.update(dict(self.runner).get("runner_parameters", {}))
+        secret_parameters = get_secret_parameters(parameters=parameters)
+        encrpyted_parameters = encrypt_secret_parameters(
+            self.parameters, secret_parameters, self.encryption_key
+        )
+        self.parameters = encrpyted_parameters
+        liveaction_dict = dict(self.liveaction)
+        if "parameters" in liveaction_dict:
+            # We need to also encrypt the parameters inside liveaction
+            original_liveaction_parameters = liveaction_dict.get("parameters", {})
+            encrpyted_parameters = encrypt_secret_parameters(
+                original_liveaction_parameters, secret_parameters, self.encryption_key
+            )
+            liveaction_dict["parameters"] = encrpyted_parameters
+            self.liveaction = liveaction_dict
+            # We also mask response found inside parameters under liveaction.
+            # As mentioned above in mask_secrets function but I don't know what should be
+            # the expected behaviour as there we are making all the values because
+            # the schema is unknown
+        original_output_value = None
+        if self.result:
+            original_output_value = self.result
+            schema = dict(self.action).get("output_schema")
+            if schema is not None:
+                self.result = output_schema.encrypt_secret_output(
+                    self.encryption_key, self.result, schema
+                )
+                # # Need output key
+                # schema = self.action.get("output_schema")
+                # for key, spec in schema.items():
+                #     if key in self.result and spec.get("secret", False):
+                #         self.result[key] = str(symmetric_encrypt(self.encryption_key, self.result[key]))
+
+        self = super(ActionExecutionDB, self).save(*args, **kwargs)
+        # Resetting to the original values
+        setattr(self, "parameters", original_parameters)
+        if hasattr(self, "liveaction"):
+            liveaction_dict = dict(self.liveaction)
+            if "parameters" in liveaction_dict:
+                liveaction_dict["parameters"] = original_liveaction_parameters
+            self.liveaction = liveaction_dict
+
+        if hasattr(self, "result") and original_output_value is not None:
+            setattr(self, "result", original_output_value)
+        return self
+
+    def update(self, **kwargs):
+        parameters = {}
+        parameters.update(dict(self.action).get("parameters", {}))
+        parameters.update(dict(self.runner).get("runner_parameters", {}))
+        secret_parameters = get_secret_parameters(parameters=parameters)
+        encrpyted_parameters = encrypt_secret_parameters(
+            self.parameters, secret_parameters, self.encryption_key
+        )
+        self.parameters = encrpyted_parameters
+        if "set__liveaction" in kwargs and "parameters" in kwargs["set__liveaction"]:
+            encrpyted_parameters = encrypt_secret_parameters(
+                kwargs["set__liveaction"]["parameters"],
+                secret_parameters,
+                self.encryption_key,
+            )
+            kwargs["set__liveaction"]["parameters"] = encrpyted_parameters
+        if "set__result" in kwargs and "result" in kwargs["set__result"]:
+            output_value = kwargs["set__result"]["result"]
+            # Need output key
+            schema = dict(self.action).get("output_schema")
+            kwargs["set__result"]["result"] = output_schema.encrypt_secret_output(
+                self.encryption_key, output_value, schema
+            )
+
+        if "parameters" in self.liveaction:
+            original_liveaction_parameters = self.liveaction.get("parameters", {})
+            encrpyted_parameters = encrypt_secret_parameters(
+                original_liveaction_parameters, secret_parameters, self.encryption_key
+            )
+            self.liveaction.parameters = encrpyted_parameters
+
+        return super(ActionExecutionDB, self).update(**kwargs)
+
 
 class ActionExecutionOutputDB(stormbase.StormFoundationDB):
     """
diff --git a/st2common/st2common/models/db/liveaction.py b/st2common/st2common/models/db/liveaction.py
index aef52462a6..5a3762f382 100644
--- a/st2common/st2common/models/db/liveaction.py
+++ b/st2common/st2common/models/db/liveaction.py
@@ -25,8 +25,11 @@
 from st2common.fields import ComplexDateTimeField
 from st2common.fields import JSONDictEscapedFieldCompatibilityField
 from st2common.util import date as date_utils
+from oslo_config import cfg
 from st2common.util.secrets import get_secret_parameters
 from st2common.util.secrets import mask_secret_parameters
+from st2common.util.secrets import encrypt_secret_parameters
+from st2common.util.crypto import read_crypto_key
 
 __all__ = [
     "LiveActionDB",
@@ -94,6 +97,7 @@ class LiveActionDB(stormbase.StormFoundationDB):
             {"fields": ["task_execution"]},
         ],
     }
+    encryption_key = read_crypto_key(cfg.CONF.actionrunner.encryption_key_path)
 
     def mask_secrets(self, value):
         from st2common.util import action_db
@@ -125,6 +129,29 @@ def get_masked_parameters(self):
         serializable_dict = self.to_serializable_dict(mask_secrets=True)
         return serializable_dict["parameters"]
 
+    def save(self, *args, **kwargs):
+        from st2common.util import action_db
+
+        original_parameters = copy.deepcopy(self.parameters)
+        action_parameters = action_db.get_action_parameters_specs(
+            action_ref=self.action
+        )
+        secret_parameters = get_secret_parameters(parameters=action_parameters)
+        encrpyted_parameters = encrypt_secret_parameters(
+            self.parameters, secret_parameters, self.encryption_key
+        )
+        self.parameters = encrpyted_parameters
+        self = super(LiveActionDB, self).save(*args, **kwargs)
+        # Reset parameters to original value after saving them to mongo
+        if "parameters" in self:
+            self.parameters = original_parameters
+        return self
+
+    def update(self, **kwargs):
+        # TODO : As of now update is not being used for LiveAction and that is why we did not
+        # add the encryption logic here but we should add it here soon
+        return super(LiveActionDB, self).update(**kwargs)
+
 
 # specialized access objects
 liveaction_access = MongoDBAccess(LiveActionDB)
diff --git a/st2common/st2common/persistence/execution.py b/st2common/st2common/persistence/execution.py
index 2073dda17b..40b65cbf8c 100644
--- a/st2common/st2common/persistence/execution.py
+++ b/st2common/st2common/persistence/execution.py
@@ -19,6 +19,10 @@
 from st2common.models.db.execution import ActionExecutionDB
 from st2common.models.db.execution import ActionExecutionOutputDB
 from st2common.persistence.base import Access
+from oslo_config import cfg
+from st2common.util.crypto import read_crypto_key
+from st2common.util.secrets import get_secret_parameters
+from st2common.util.secrets import decrypt_secret_parameters
 
 __all__ = [
     "ActionExecution",
@@ -29,6 +33,7 @@
 class ActionExecution(Access):
     impl = MongoDBAccess(ActionExecutionDB)
     publisher = None
+    encryption_key = read_crypto_key(cfg.CONF.actionrunner.encryption_key_path)
 
     @classmethod
     def _get_impl(cls):
@@ -44,6 +49,66 @@ def _get_publisher(cls):
     def delete_by_query(cls, *args, **query):
         return cls._get_impl().delete_by_query(*args, **query)
 
+    @classmethod
+    def get_by_name(cls, value):
+        result = cls.get(name=value, raise_exception=True)
+        return result
+
+    @classmethod
+    def get_by_id(cls, value):
+        instance = super(ActionExecution, cls).get_by_id(value)
+        instance = cls._decrypt_secrets(instance)
+        return instance
+
+    @classmethod
+    def get_by_uid(cls, value):
+        result = cls.get(uid=value, raise_exception=True)
+        return result
+
+    @classmethod
+    def get_by_ref(cls, value):
+        result = cls.get(ref=value, raise_exception=True)
+        return result
+
+    @classmethod
+    def get_by_pack(cls, value):
+        result = cls.get(pack=value, raise_exception=True)
+        return result
+
+    @classmethod
+    def get(cls, *args, **kwargs):
+        instance = super(ActionExecution, cls).get(*args, **kwargs)
+        if instance is None:
+            return instance
+        # Decrypt secrets if any
+        instance = cls._decrypt_secrets(instance)
+        return instance
+
+    @classmethod
+    def _decrypt_secrets(cls, instance):
+        if instance is None:
+            return instance
+        action = getattr(instance, "action", {})
+        runner = getattr(instance, "runner", {})
+        parameters = {}
+        parameters.update(action.get("parameters", {}))
+        parameters.update(runner.get("runner_parameters", {}))
+        secret_parameters = get_secret_parameters(parameters=parameters)
+
+        decrypt_parameters = decrypt_secret_parameters(
+            getattr(instance, "parameters", {}), secret_parameters, cls.encryption_key
+        )
+        setattr(instance, "parameters", decrypt_parameters)
+
+        liveaction_parameter = getattr(instance, "liveaction", {}).get("parameters", {})
+        if liveaction_parameter:
+            decrypt_liveaction_parameters = decrypt_secret_parameters(
+                liveaction_parameter, secret_parameters, cls.encryption_key
+            )
+            instance.liveaction["parameters"] = decrypt_liveaction_parameters
+
+        return instance
+
 
 class ActionExecutionOutput(Access):
     impl = MongoDBAccess(ActionExecutionOutputDB)
diff --git a/st2common/st2common/persistence/liveaction.py b/st2common/st2common/persistence/liveaction.py
index aa7551592a..a827a02989 100644
--- a/st2common/st2common/persistence/liveaction.py
+++ b/st2common/st2common/persistence/liveaction.py
@@ -18,6 +18,9 @@
 from st2common import transport
 from st2common.models.db.liveaction import liveaction_access
 from st2common.persistence import base as persistence
+from oslo_config import cfg
+from st2common.util.crypto import read_crypto_key
+from st2common.util.secrets import decrypt_secret_parameters, get_secret_parameters
 
 __all__ = ["LiveAction"]
 
@@ -25,6 +28,7 @@
 class LiveAction(persistence.StatusBasedResource):
     impl = liveaction_access
     publisher = None
+    encryption_key = read_crypto_key(cfg.CONF.actionrunner.encryption_key_path)
 
     @classmethod
     def _get_impl(cls):
@@ -39,3 +43,46 @@ def _get_publisher(cls):
     @classmethod
     def delete_by_query(cls, *args, **query):
         return cls._get_impl().delete_by_query(*args, **query)
+
+    @classmethod
+    def get(self, *args, **kwargs):
+        return super(LiveAction, self).get(*args, **kwargs)
+
+    @classmethod
+    def get_by_id(cls, value):
+        from st2common.util import action_db
+
+        instance = super(LiveAction, cls).get_by_id(value)
+        parameters = getattr(instance, "parameters", None)
+        action = getattr(instance, "action", None)
+        action_parameters = action_db.get_action_parameters_specs(action_ref=action)
+        secret_parameters = get_secret_parameters(parameters=action_parameters)
+        decrypt_parameters = decrypt_secret_parameters(
+            parameters, secret_parameters, cls.encryption_key
+        )
+        setattr(instance, "parameters", decrypt_parameters)
+        return instance
+
+    @classmethod
+    def get_by_name(cls, value):
+        # TODO: Ideally we should add decryption logic here after getting the data from mongo DB
+        instance = super(LiveAction, cls).get_by_name(value)
+        return instance
+
+    @classmethod
+    def get_by_uid(cls, value):
+        # TODO: Ideally we should add decryption logic here after getting the data from mongo DB
+        instance = super(LiveAction, cls).get_by_uid(value)
+        return instance
+
+    @classmethod
+    def get_by_ref(cls, value):
+        # TODO: Ideally we should add decryption logic here after getting the data from mongo DB
+        instance = super(LiveAction, cls).get_by_ref(value)
+        return instance
+
+    @classmethod
+    def get_by_pack(cls, value):
+        # TODO: Ideally we should add decryption logic here after getting the data from mongo DB
+        instance = super(LiveAction, cls).get_by_pack(value)
+        return instance
diff --git a/st2common/st2common/services/executions.py b/st2common/st2common/services/executions.py
index 80706e8f79..54ccc937f1 100644
--- a/st2common/st2common/services/executions.py
+++ b/st2common/st2common/services/executions.py
@@ -149,6 +149,13 @@ def create_execution_object(
 
     # TODO: This object initialization takes 20-30or so ms
     execution = ActionExecutionDB(**attrs)
+    LOG.debug(
+        "Execution : create_execution_object : liveaction : %s", execution.liveaction
+    )
+    LOG.debug(
+        "Execution : create_execution_object : liveaction : type : %s",
+        type(execution.liveaction),
+    )
     # TODO: Do 100% research this is fully safe and unique in distributed setups
     execution.id = ObjectId()
     execution.web_url = _get_web_url_for_execution(str(execution.id))
@@ -158,7 +165,6 @@ def create_execution_object(
     execution = ActionExecution.add_or_update(
         execution, publish=publish, validate=False
     )
-
     if parent and str(execution.id) not in parent.children:
         values = {}
         values["push__children"] = str(execution.id)
@@ -195,7 +201,6 @@ def update_execution(liveaction_db, publish=True, set_result_size=False):
                             on the "result_size" database field.
     """
     execution = ActionExecution.get(liveaction__id=str(liveaction_db.id))
-
     with coordination.get_coordinator().get_lock(str(liveaction_db.id).encode()):
         # Skip execution object update when action is already in completed state.
         if execution.status in action_constants.LIVEACTION_COMPLETED_STATES:
diff --git a/st2common/st2common/util/action_db.py b/st2common/st2common/util/action_db.py
index e6ae0fe430..99a23c2796 100644
--- a/st2common/st2common/util/action_db.py
+++ b/st2common/st2common/util/action_db.py
@@ -75,8 +75,13 @@ def get_action_parameters_specs(action_ref):
     runner_type_db = get_runnertype_by_name(runnertype_name=runner_type_name)
 
     # Runner type parameters should be added first before the action parameters.
-    parameters.update(runner_type_db["runner_parameters"])
-    parameters.update(action_db.parameters)
+    runner_parameters = getattr(runner_type_db, "runner_parameters")
+    action_parameters = getattr(action_db, "parameters")
+    # We need to check if the runner parameters are None or not before updating parameters
+    if runner_parameters is not None:
+        parameters.update(runner_type_db["runner_parameters"])
+    if action_parameters is not None:
+        parameters.update(action_db.parameters)
 
     return parameters
 
diff --git a/st2common/st2common/util/crypto.py b/st2common/st2common/util/crypto.py
index 0aea24763c..0bae0e7d1e 100644
--- a/st2common/st2common/util/crypto.py
+++ b/st2common/st2common/util/crypto.py
@@ -275,7 +275,7 @@ def cryptography_symmetric_encrypt(encrypt_key, plaintext):
     result = msg_bytes + sig_bytes
 
     # Convert resulting byte string to hex notation ASCII string
-    result = binascii.hexlify(result).upper()
+    result = binascii.hexlify(result).upper().decode()
 
     return result
 
diff --git a/st2common/st2common/util/output_schema.py b/st2common/st2common/util/output_schema.py
index 902908b653..eda78328c6 100644
--- a/st2common/st2common/util/output_schema.py
+++ b/st2common/st2common/util/output_schema.py
@@ -23,6 +23,7 @@
 from st2common.util import schema
 from st2common.constants import action as action_constants
 from st2common.constants.secrets import MASKED_ATTRIBUTE_VALUE
+from st2common.util.crypto import symmetric_encrypt
 
 
 LOG = logging.getLogger(__name__)
@@ -195,6 +196,15 @@ def mask_secret_output(ac_ex, output_value):
     return output_value
 
 
+def encrypt_secret_output(encryption_key, output_value, output_schema):
+    for key, spec in output_schema.items():
+        if key in output_value and spec.get("secret", False):
+            output_value[key] = str(
+                symmetric_encrypt(encryption_key, output_value[key])
+            )
+    return output_value
+
+
 def validate_output(runner_schema, action_schema, result, status, output_key):
     """Validate output of action with runner and action schema."""
     try:
diff --git a/st2common/st2common/util/secrets.py b/st2common/st2common/util/secrets.py
index 850152a5fa..36ca9f52a4 100644
--- a/st2common/st2common/util/secrets.py
+++ b/st2common/st2common/util/secrets.py
@@ -22,6 +22,7 @@
 import six
 
 from st2common.util.deep_copy import fast_deepcopy_dict
+from st2common.util.crypto import symmetric_encrypt, symmetric_decrypt
 from st2common.constants.secrets import MASKED_ATTRIBUTE_VALUE
 
 
@@ -136,6 +137,113 @@ def get_secret_parameters(parameters):
     return secret_parameters
 
 
+def decrypt_secret_parameters(
+    parameters, secret_parameters, encryption_key, result=None
+):
+    iterator = None
+    is_dict = isinstance(secret_parameters, dict)
+    is_list = isinstance(secret_parameters, list)
+    if is_dict:
+        iterator = six.iteritems(secret_parameters)
+    elif is_list:
+        iterator = enumerate(secret_parameters)
+    else:
+        return str(symmetric_decrypt(encryption_key, parameters))
+
+    # only create a deep copy of parameters on the first call
+    # all other recursive calls pass back referneces to this result object
+    # so we can reuse it, saving memory and CPU cycles
+    if result is None:
+        result = fast_deepcopy_dict(parameters)
+
+    # iterate over the secret parameters
+    for secret_param, secret_sub_params in iterator:
+        if is_dict:
+            if secret_param in result:
+                result[secret_param] = decrypt_secret_parameters(
+                    parameters[secret_param],
+                    secret_sub_params,
+                    encryption_key,
+                    result=result[secret_param],
+                )
+        elif is_list:
+            # we're assuming lists contain the same data type for every element
+            for idx, value in enumerate(result):
+                result[idx] = decrypt_secret_parameters(
+                    parameters[idx],
+                    secret_sub_params,
+                    encryption_key,
+                    result=result[idx],
+                )
+        else:
+            result[secret_param] = symmetric_decrypt(encryption_key, parameters)
+
+    return result
+
+
+def encrypt_secret_parameters(
+    parameters, secret_parameters, encryption_key, result=None
+):
+    """
+    Introspect the parameters dict and return a new dict with encyrpted secret
+    parameters.
+    :param parameters: Parameters to process.
+    :type parameters: ``dict`` or ``list`` or ``string``
+
+    :param secret_parameters: Dict of parameter names which are secret.
+                              The type must be the same type as ``parameters``
+                              (or at least behave in the same way),
+                              so that they can be traversed in the same way as
+                              recurse down into the structure.
+    :type secret_parameters: ``dict``
+
+    :param result: Deep copy of parameters so that parameters is not modified
+                   in place. Default = None, meaning this function will make a
+                   deep copy before starting.
+    :type result: ``dict`` or ``list`` or ``string``
+    """
+    # how we iterate depends on what data type was passed in
+    iterator = None
+    is_dict = isinstance(secret_parameters, dict)
+    is_list = isinstance(secret_parameters, list)
+    if is_dict:
+        iterator = six.iteritems(secret_parameters)
+    elif is_list:
+        iterator = enumerate(secret_parameters)
+    else:
+        return symmetric_encrypt(encryption_key, parameters)
+
+    # only create a deep copy of parameters on the first call
+    # all other recursive calls pass back referneces to this result object
+    # so we can reuse it, saving memory and CPU cycles
+    if result is None:
+        result = fast_deepcopy_dict(parameters)
+
+    # iterate over the secret parameters
+    for secret_param, secret_sub_params in iterator:
+        if is_dict:
+            if secret_param in result:
+                result[secret_param] = encrypt_secret_parameters(
+                    parameters[secret_param],
+                    secret_sub_params,
+                    encryption_key,
+                    result=result[secret_param],
+                )
+        elif is_list:
+            # we're assuming lists contain the same data type for every element
+            for idx, value in enumerate(result):
+                result[idx] = encrypt_secret_parameters(
+                    parameters[idx],
+                    secret_sub_params,
+                    encryption_key,
+                    result=result[idx],
+                )
+        else:
+            result[secret_param] = symmetric_encrypt(encryption_key, parameters)
+
+    return result
+
+
 def mask_secret_parameters(parameters, secret_parameters, result=None):
     """
     Introspect the parameters dict and return a new dict with masked secret