openstack/octavia
Code Issues Proposed changes

Merge "Updating ssh driver with root user check"

Browse Source
This commit is contained in:
Jenkins
2015-08-18 22:22:24 +00:00
committed by Gerrit Code Review
parent 4f996d000a fe9f506a07
commit 71e5725a7f
2 changed files with 18 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
octavia/amphorae/drivers/haproxy/ssh_driver.py
View File

@@ -208,7 +208,7 @@ class HaproxyManager(driver_base.AmphoraLoadBalancerDriver):
self.client.close()
def _execute_command(self, command, run_as_root=False):
if run_as_root:
if run_as_root and not self._is_root():
command = "sudo {0}".format(command)
_, stdout, stderr = self.client.exec_command(command)
stdout = stdout.read()
@@ -319,3 +319,6 @@ class HaproxyManager(driver_base.AmphoraLoadBalancerDriver):
# Close the temp file
for temp in temps:
temp.close()
def _is_root(self):
return cfg.CONF.haproxy_amphora.username == 'root'

21
octavia/tests/unit/amphorae/drivers/haproxy/test_ssh_driver.py
View File

@@ -12,6 +12,7 @@
# License for the specific language governing permissions and limitations
# under the License.
from oslo_config import cfg
from oslo_log import log
from oslo_utils import uuidutils
import paramiko
@@ -207,6 +208,13 @@ class TestSshDriver(base.TestCase):
mock.call(mock.ANY, mock.ANY),
])
def test_build_pem(self):
expected = 'imainter\nimainter2\nimacert\nimakey'
tls_tupe = sample_configs.sample_tls_container_tuple(
certificate='imacert', private_key='imakey',
intermediates=['imainter', 'imainter2'])
self.assertEqual(expected, cert_parser.build_pem(tls_tupe))
def test_get_primary_cn(self):
cert = mock.MagicMock()
@@ -240,13 +248,6 @@ class TestSshDriver(base.TestCase):
self.driver._map_cert_tls_container(
cert).intermediates)
def test_build_pem(self):
expected = 'imainter\nimainter2\nimacert\nimakey'
tls_tupe = sample_configs.sample_tls_container_tuple(
certificate='imacert', private_key='imakey',
intermediates=['imainter', 'imainter2'])
self.assertEqual(expected, cert_parser.build_pem(tls_tupe))
@mock.patch.object(ssh_driver.HaproxyManager, '_execute_command')
def test_post_vip_plug_no_down_links(self, exec_command):
amps = [data_models.Amphora(id=MOCK_AMP_ID, compute_id=MOCK_COMPUTE_ID,
@@ -335,3 +336,9 @@ class TestSshDriver(base.TestCase):
show_ip_call = mock.call(ssh_driver.CMD_SHOW_IP_ADDR.format(iface))
exec_command.assert_has_calls([grep_call, dhclient_call, show_ip_call])
self.assertEqual(3, exec_command.call_count)
def test_is_root(self):
cfg.CONF.set_override('username', 'root', group='haproxy_amphora')
self.assertTrue(self.driver._is_root())
cfg.CONF.set_override('username', 'blah', group='haproxy_amphora')
self.assertFalse(self.driver._is_root())