Fix unicode bug for password-authenticated BGP peer

Using encodeutils convert password to utf-8 format.

Change-Id: I77211a75e38d8dcc8cfa54e1c31a3b94c277c3a8
Closes-Bug: #1592982
This commit is contained in:
Dongcan Ye 2016-07-05 18:04:41 +08:00
parent e2581d50e7
commit 15c83c5b12
2 changed files with 32 additions and 4 deletions

View File

@ -14,6 +14,7 @@
# under the License.
from oslo_log import log as logging
from oslo_utils import encodeutils
from ryu.services.protocols.bgp import bgpspeaker
from ryu.services.protocols.bgp.rtconf.neighbors import CONNECT_MODE_ACTIVE
@ -119,6 +120,8 @@ class RyuBgpDriver(base.BgpDriverBase):
utils.validate_as_num('remote_as', peer_as)
utils.validate_string(peer_ip)
utils.validate_auth(auth_type, password)
if password is not None:
password = encodeutils.to_utf8(password)
# Notify Ryu about BGP Peer addition
curr_speaker.neighbor_add(address=peer_ip,

View File

@ -13,8 +13,11 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import six
import mock
from oslo_config import cfg
from oslo_utils import encodeutils
from ryu.services.protocols.bgp import bgpspeaker
from ryu.services.protocols.bgp.rtconf.neighbors import CONNECT_MODE_ACTIVE
@ -98,7 +101,29 @@ class TestRyuBgpDriver(base.BaseTestCase):
speaker.neighbor_add.assert_called_once_with(
address=FAKE_PEER_IP,
remote_as=FAKE_PEER_AS,
password=FAKE_PEER_PASSWORD,
password=encodeutils.to_utf8(FAKE_PEER_PASSWORD),
connect_mode=CONNECT_MODE_ACTIVE)
def test_add_bgp_peer_with_unicode_password(self):
self.ryu_bgp_driver.add_bgp_speaker(FAKE_LOCAL_AS1)
self.assertEqual(1,
self.ryu_bgp_driver.cache.get_hosted_bgp_speakers_count())
# In Python3 a str is unicode
if six.PY3:
NEW_FAKE_PEER_PASSWORD = str(FAKE_PEER_PASSWORD)
else:
NEW_FAKE_PEER_PASSWORD = unicode(FAKE_PEER_PASSWORD)
self.ryu_bgp_driver.add_bgp_peer(
FAKE_LOCAL_AS1,
FAKE_PEER_IP,
FAKE_PEER_AS,
FAKE_AUTH_TYPE,
NEW_FAKE_PEER_PASSWORD)
speaker = self.ryu_bgp_driver.cache.get_bgp_speaker(FAKE_LOCAL_AS1)
speaker.neighbor_add.assert_called_once_with(
address=FAKE_PEER_IP,
remote_as=FAKE_PEER_AS,
password=encodeutils.to_utf8(NEW_FAKE_PEER_PASSWORD),
connect_mode=CONNECT_MODE_ACTIVE)
def test_remove_bgp_peer(self):