1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290
|
# Copyright (C) 2007-2010 Canonical Ltd
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
from bzrlib.tests import TestCase
from bzrlib.errors import SSHVendorNotFound, UnknownSSH
from bzrlib.transport.ssh import (
OpenSSHSubprocessVendor,
PLinkSubprocessVendor,
SSHCorpSubprocessVendor,
LSHSubprocessVendor,
SSHVendorManager,
StrangeHostname,
)
class TestSSHVendorManager(SSHVendorManager):
_ssh_version_string = ""
def set_ssh_version_string(self, version):
self._ssh_version_string = version
def _get_ssh_version_string(self, args):
return self._ssh_version_string
class SSHVendorManagerTests(TestCase):
def test_register_vendor(self):
manager = TestSSHVendorManager()
self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
vendor = object()
manager.register_vendor("vendor", vendor)
self.assertIs(manager.get_vendor({"BZR_SSH": "vendor"}), vendor)
def test_default_vendor(self):
manager = TestSSHVendorManager()
self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
vendor = object()
manager.register_default_vendor(vendor)
self.assertIs(manager.get_vendor({}), vendor)
def test_get_vendor_by_environment(self):
manager = TestSSHVendorManager()
self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
self.assertRaises(UnknownSSH,
manager.get_vendor, {"BZR_SSH": "vendor"})
vendor = object()
manager.register_vendor("vendor", vendor)
self.assertIs(manager.get_vendor({"BZR_SSH": "vendor"}), vendor)
def test_get_vendor_by_inspection_openssh(self):
manager = TestSSHVendorManager()
self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
manager.set_ssh_version_string("OpenSSH")
self.assertIsInstance(manager.get_vendor({}), OpenSSHSubprocessVendor)
def test_get_vendor_by_inspection_sshcorp(self):
manager = TestSSHVendorManager()
self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
manager.set_ssh_version_string("SSH Secure Shell")
self.assertIsInstance(manager.get_vendor({}), SSHCorpSubprocessVendor)
def test_get_vendor_by_inspection_lsh(self):
manager = TestSSHVendorManager()
self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
manager.set_ssh_version_string("lsh")
self.assertIsInstance(manager.get_vendor({}), LSHSubprocessVendor)
def test_get_vendor_by_inspection_plink(self):
manager = TestSSHVendorManager()
self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
manager.set_ssh_version_string("plink")
# Auto-detect of plink vendor disabled, on Windows recommended
# default ssh-client is paramiko
# see https://bugs.launchpad.net/bugs/414743
#~self.assertIsInstance(manager.get_vendor({}), PLinkSubprocessVendor)
self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
def test_cached_vendor(self):
manager = TestSSHVendorManager()
self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
vendor = object()
manager.register_vendor("vendor", vendor)
self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
# Once the vendor is found the result is cached (mainly because of the
# 'get_vendor' sometimes can be an expensive operation) and later
# invocations of the 'get_vendor' just returns the cached value.
self.assertIs(manager.get_vendor({"BZR_SSH": "vendor"}), vendor)
self.assertIs(manager.get_vendor({}), vendor)
# The cache can be cleared by the 'clear_cache' method
manager.clear_cache()
self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
def test_get_vendor_search_order(self):
# The 'get_vendor' method search for SSH vendors as following:
#
# 1. Check previously cached value
# 2. Check BZR_SSH environment variable
# 3. Check the system for known SSH vendors
# 4. Fall back to the default vendor if registered
#
# Let's now check the each check method in the reverse order
# clearing the cache between each invocation:
manager = TestSSHVendorManager()
# At first no vendors are found
self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
# If the default vendor is registered it will be returned
default_vendor = object()
manager.register_default_vendor(default_vendor)
self.assertIs(manager.get_vendor({}), default_vendor)
# If the known vendor is found in the system it will be returned
manager.clear_cache()
manager.set_ssh_version_string("OpenSSH")
self.assertIsInstance(manager.get_vendor({}), OpenSSHSubprocessVendor)
# If the BZR_SSH environment variable is found it will be treated as
# the vendor name
manager.clear_cache()
vendor = object()
manager.register_vendor("vendor", vendor)
self.assertIs(manager.get_vendor({"BZR_SSH": "vendor"}), vendor)
# Last cached value always checked first
self.assertIs(manager.get_vendor({}), vendor)
def test_get_vendor_from_path_win32_plink(self):
manager = TestSSHVendorManager()
manager.set_ssh_version_string("plink: Release 0.60")
plink_path = "C:/Program Files/PuTTY/plink.exe"
vendor = manager.get_vendor({"BZR_SSH": plink_path})
self.assertIsInstance(vendor, PLinkSubprocessVendor)
args = vendor._get_vendor_specific_argv("user", "host", 22, ["bzr"])
self.assertEqual(args[0], plink_path)
def test_get_vendor_from_path_nix_openssh(self):
manager = TestSSHVendorManager()
manager.set_ssh_version_string(
"OpenSSH_5.1p1 Debian-5, OpenSSL, 0.9.8g 19 Oct 2007")
openssh_path = "/usr/bin/ssh"
vendor = manager.get_vendor({"BZR_SSH": openssh_path})
self.assertIsInstance(vendor, OpenSSHSubprocessVendor)
args = vendor._get_vendor_specific_argv("user", "host", 22, ["bzr"])
self.assertEqual(args[0], openssh_path)
class SubprocessVendorsTests(TestCase):
def test_openssh_command_tricked(self):
vendor = OpenSSHSubprocessVendor()
self.assertEqual(
vendor._get_vendor_specific_argv(
"user", "-oProxyCommand=blah", 100, command=["bzr"]),
["ssh", "-oForwardX11=no", "-oForwardAgent=no",
"-oClearAllForwardings=yes",
"-oNoHostAuthenticationForLocalhost=yes",
"-p", "100",
"-l", "user",
"--",
"-oProxyCommand=blah", "bzr"])
def test_openssh_command_arguments(self):
vendor = OpenSSHSubprocessVendor()
self.assertEqual(
vendor._get_vendor_specific_argv(
"user", "host", 100, command=["bzr"]),
["ssh", "-oForwardX11=no", "-oForwardAgent=no",
"-oClearAllForwardings=yes",
"-oNoHostAuthenticationForLocalhost=yes",
"-p", "100",
"-l", "user",
"--",
"host", "bzr"]
)
def test_openssh_subsystem_arguments(self):
vendor = OpenSSHSubprocessVendor()
self.assertEqual(
vendor._get_vendor_specific_argv(
"user", "host", 100, subsystem="sftp"),
["ssh", "-oForwardX11=no", "-oForwardAgent=no",
"-oClearAllForwardings=yes",
"-oNoHostAuthenticationForLocalhost=yes",
"-p", "100",
"-l", "user",
"-s", "--", "host", "sftp"]
)
def test_openssh_command_tricked(self):
vendor = SSHCorpSubprocessVendor()
self.assertRaises(
StrangeHostname,
vendor._get_vendor_specific_argv,
"user", "-oProxyCommand=host", 100, command=["bzr"])
def test_sshcorp_command_arguments(self):
vendor = SSHCorpSubprocessVendor()
self.assertEqual(
vendor._get_vendor_specific_argv(
"user", "host", 100, command=["bzr"]),
["ssh", "-x",
"-p", "100",
"-l", "user",
"host", "bzr"]
)
def test_sshcorp_subsystem_arguments(self):
vendor = SSHCorpSubprocessVendor()
self.assertEqual(
vendor._get_vendor_specific_argv(
"user", "host", 100, subsystem="sftp"),
["ssh", "-x",
"-p", "100",
"-l", "user",
"-s", "sftp", "host"]
)
def test_lsh_command_tricked(self):
vendor = LSHSubprocessVendor()
self.assertRaises(
StrangeHostname,
vendor._get_vendor_specific_argv,
"user", "-oProxyCommand=host", 100, command=["bzr"])
def test_lsh_command_arguments(self):
vendor = LSHSubprocessVendor()
self.assertEqual(
vendor._get_vendor_specific_argv(
"user", "host", 100, command=["bzr"]),
["lsh",
"-p", "100",
"-l", "user",
"host", "bzr"]
)
def test_lsh_subsystem_arguments(self):
vendor = LSHSubprocessVendor()
self.assertEqual(
vendor._get_vendor_specific_argv(
"user", "host", 100, subsystem="sftp"),
["lsh",
"-p", "100",
"-l", "user",
"--subsystem", "sftp", "host"]
)
def test_plink_command_tricked(self):
vendor = PLinkSubprocessVendor()
self.assertRaises(
StrangeHostname,
vendor._get_vendor_specific_argv,
"user", "-oProxyCommand=host", 100, command=["bzr"])
def test_plink_command_arguments(self):
vendor = PLinkSubprocessVendor()
self.assertEqual(
vendor._get_vendor_specific_argv(
"user", "host", 100, command=["bzr"]),
["plink", "-x", "-a", "-ssh", "-2", "-batch",
"-P", "100",
"-l", "user",
"host", "bzr"]
)
def test_plink_subsystem_arguments(self):
vendor = PLinkSubprocessVendor()
self.assertEqual(
vendor._get_vendor_specific_argv(
"user", "host", 100, subsystem="sftp"),
["plink", "-x", "-a", "-ssh", "-2", "-batch",
"-P", "100",
"-l", "user",
"-s", "host", "sftp"]
)
|