gpg_cmd is not allowed as plugin or override configuration

Since the command configured with gpg_cmd executes remotely as user apache,
a user should not be allowed to change it via a distributor config or
an override at publish time.

Fixes #3498
https://pulp.plan.io/issues/3498

Change-Id: I88cdb4f51c237b1157e7424863df7049269939ca
(cherry picked from commit 1c51268)

plugins/pulp_deb/plugins/distributors/configuration.py

@@ -18,6 +18,8 @@
                         PUBLISH_DEFAULT_RELEASE_KEYWORD,
                         GPG_CMD, GPG_KEY_ID)
 
+LOCAL_CONFIG_KEYS = [GPG_CMD]
+
 ROOT_PUBLISH_DIR = '/var/lib/pulp/published/deb'
 MASTER_PUBLISH_DIR = os.path.join(ROOT_PUBLISH_DIR, 'master')
 HTTP_PUBLISH_DIR = os.path.join(ROOT_PUBLISH_DIR, 'http', 'repos')
@@ -40,10 +42,22 @@ def validate_config(repo, config, config_conduit):
     or not and why
     :rtype:  tuple of (bool, str or None)
     """
+    # Keys in LOCAL_CONFIG_KEYS cannot be set in remote configs
+    # Perform these the checks before flattening the config, to
+    # give feedback on which configuration carries invalid options
+    error_messages = []
+    msg = _('Configuration key [%(k)s] is not allowed in %(config)s configuration')
+    remote_configs = [
+        (config.repo_plugin_config, "repository plugin"),
+        (config.override_config, "override")]
+    for key in LOCAL_CONFIG_KEYS:
+        for cfgdict, cfgname in remote_configs:
+            if cfgdict.get(key):
+                error_messages.append(msg % dict(k=key, config=cfgname))
+
     # squish it into a dictionary so we can manipulate it
     if not isinstance(config, dict):
         config = config.flatten()
-    error_messages = []
 
     configured_keys = set(config)
     required_keys = set(REQUIRED_CONFIG_KEYS)

plugins/test/unit/plugins/distributors/test_configuration.py

@@ -95,3 +95,15 @@ def test_create_distributor_same_url(self):
                            'conflicts with existing relative URL [/bar] ' +
                            'for repository [foo]'),
                           configuration.validate_config(repo, config, conduit))
+
+    def test__repocfg_gpg_cmd(self):
+        config = PluginCallConfiguration(
+            dict(http=True, https=False, relative_url='fool'),
+            dict(gpg_cmd="/bin/true should fail"))
+        repo = Mock(repo_id='fool', working_dir=self.work_dir)
+        conduit = self._config_conduit()
+
+        expected_reason = ('Configuration key [gpg_cmd] is not allowed '
+                           'in repository plugin configuration')
+        self.assertEquals((False, expected_reason),
+                          configuration.validate_config(repo, config, conduit))

plugins/test/unit/plugins/distributors/test_distributor.py

@@ -8,6 +8,7 @@
 import mock
 from .... import testbase
 
+from pulp.plugins.config import PluginCallConfiguration
 from pulp_deb.common import ids, constants
 from pulp_deb.plugins.db import models
 
@@ -59,7 +60,7 @@ class TestConfiguration(BaseTest):
     def test_validate_config_empty(self):
         repo = mock.MagicMock(id="repo-1")
         conduit = self._config_conduit()
-        config = {}
+        config = PluginCallConfiguration({}, {})
         distributor = self.Module.DebDistributor()
         self.assertEquals(
             (False, '\n'.join([
@@ -76,8 +77,9 @@ def test_validate_config(self):
 
         repo = mock.MagicMock(id="repo-1")
         conduit = self._config_conduit()
-        config = dict(http=True, https=False, relative_url=None,
-                      gpg_cmd=signer)
+        config = PluginCallConfiguration(
+            dict(gpg_cmd=signer),
+            dict(http=True, https=False, relative_url=None))
         distributor = self.Module.DebDistributor()
         self.assertEquals(
             distributor.validate_config(repo, config, conduit),
@@ -89,8 +91,9 @@ def test_validate_config_bad_signer(self):
 
         repo = mock.MagicMock(id="repo-1")
         conduit = self._config_conduit()
-        config = dict(http=True, https=False, relative_url=None,
-                      gpg_cmd=signer)
+        config = PluginCallConfiguration(
+            dict(gpg_cmd=signer),
+            dict(http=True, https=False, relative_url=None))
         distributor = self.Module.DebDistributor()
         self.assertEquals(
             (False, '\n'.join([