mirror/osrg-gobgp
1
0
Fork 0
mirror of https://github.com/osrg/gobgp.git synced 2024-05-11 05:55:10 +00:00
Code Issues Releases Activity
Files
master
osrg-gobgp/test/scenario_test/route_server_malformed_test.py
T

554 lines
15 KiB
Python
Raw Permalink Blame History

# Copyright (C) 2015 Nippon Telegraph and Telephone Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import sys
import time
import unittest
import inspect
import collections
collections.Callable = collections.abc.Callable
import nose
from lib.noseplugin import OptionParser, parser_option
from lib import base
from lib.base import BGP_FSM_ESTABLISHED, local
from lib.gobgp import GoBGPContainer
from lib.exabgp import ExaBGPContainer
counter = 1
_SCENARIOS = {}
def register_scenario(cls):
global counter
_SCENARIOS[counter] = cls
counter += 1
def lookup_scenario(name):
for value in list(_SCENARIOS.values()):
if value.__name__ == name:
return value
return None
def wait_for(f, timeout=120):
interval = 1
count = 0
while True:
if f():
return
time.sleep(interval)
count += interval
if count >= timeout:
raise Exception('timeout')
@register_scenario
class MalformedMpReachNlri(object):
"""
No.1 malformaed mp-reach-nlri
"""
@staticmethod
def boot(env):
gobgp_ctn_image_name = env.parser_option.gobgp_image
log_level = env.parser_option.gobgp_log_level
g1 = GoBGPContainer(name='g1', asn=65000, router_id='192.168.0.1',
ctn_image_name=gobgp_ctn_image_name,
log_level=log_level)
e1 = ExaBGPContainer(name='e1', asn=65001, router_id='192.168.0.2')
e2 = ExaBGPContainer(name='e2', asn=65001, router_id='192.168.0.2')
ctns = [g1, e1, e2]
initial_wait_time = max(ctn.run() for ctn in ctns)
time.sleep(initial_wait_time)
for q in [e1, e2]:
g1.add_peer(q, is_rs_client=True)
q.add_peer(g1)
env.g1 = g1
env.e1 = e1
env.e2 = e2
@staticmethod
def setup(env):
g1 = env.g1
e1 = env.e1
e2 = env.e2
for c in [e1, e2]:
g1.wait_for(BGP_FSM_ESTABLISHED, c)
# advertise malformed MP_REACH_NLRI
e1.add_route('10.7.0.17/32', attribute='0x0e 0x60 0x11223344')
@staticmethod
def check(env):
g1 = env.g1
e1 = env.e1
e2 = env.e2
def f():
for line in e1.log().split('\n'):
if 'UPDATE message error / Attribute Flags Error / 0x600E0411223344' in line:
return True
return False
wait_for(f)
# check e2 is still established
g1.wait_for(BGP_FSM_ESTABLISHED, e2)
@staticmethod
def executor(env):
lookup_scenario("MalformedMpReachNlri").boot(env)
lookup_scenario("MalformedMpReachNlri").setup(env)
lookup_scenario("MalformedMpReachNlri").check(env)
@register_scenario
class MalformedMpUnReachNlri(object):
"""
No.2 malformaed mp-unreach-nlri
"""
@staticmethod
def boot(env):
lookup_scenario("MalformedMpReachNlri").boot(env)
@staticmethod
def setup(env):
g1 = env.g1
e1 = env.e1
e2 = env.e2
for c in [e1, e2]:
g1.wait_for(BGP_FSM_ESTABLISHED, c)
# advertise malformed MP_UNREACH_NLRI
e1.add_route('10.7.0.17/32', attribute='0x0f 0x60 0x11223344')
@staticmethod
def check(env):
g1 = env.g1
e1 = env.e1
e2 = env.e2
def f():
for line in e1.log().split('\n'):
if 'UPDATE message error / Attribute Flags Error / 0x600F0411223344' in line:
return True
return False
wait_for(f)
# check e2 is still established
g1.wait_for(BGP_FSM_ESTABLISHED, e2)
@staticmethod
def executor(env):
lookup_scenario("MalformedMpUnReachNlri").boot(env)
lookup_scenario("MalformedMpUnReachNlri").setup(env)
lookup_scenario("MalformedMpUnReachNlri").check(env)
@register_scenario
class MalformedAsPath(object):
"""
No.3 malformaed as-path
"""
@staticmethod
def boot(env):
lookup_scenario("MalformedMpReachNlri").boot(env)
@staticmethod
def setup(env):
g1 = env.g1
e1 = env.e1
e2 = env.e2
for c in [e1, e2]:
g1.wait_for(BGP_FSM_ESTABLISHED, c)
# advertise malformed AS_PATH
# Send the attribute to the length and number of aspath is inconsistent
# Attribute Type 0x02 (AS_PATH)
# Attribute Flag 0x40 (well-known transitive)
# Attribute Value 0x02020000ffdc (
# segment type = 02
# segment length = 02 -> # correct value = 01
# as number = 65500 )
e1.add_route('10.7.0.17/32', attribute='0x02 0x60 0x11223344')
@staticmethod
def check(env):
g1 = env.g1
e1 = env.e1
e2 = env.e2
def f():
for line in e1.log().split('\n'):
if 'UPDATE message error / Attribute Flags Error / 0x60020411223344' in line:
return True
return False
wait_for(f)
# check e2 is still established
g1.wait_for(BGP_FSM_ESTABLISHED, e2)
@staticmethod
def executor(env):
lookup_scenario("MalformedAsPath").boot(env)
lookup_scenario("MalformedAsPath").setup(env)
lookup_scenario("MalformedAsPath").check(env)
@register_scenario
class MalformedAs4Path(object):
"""
No.4 malformaed as4-path
"""
@staticmethod
def boot(env):
lookup_scenario("MalformedMpReachNlri").boot(env)
@staticmethod
def setup(env):
g1 = env.g1
e1 = env.e1
e2 = env.e2
for c in [e1, e2]:
g1.wait_for(BGP_FSM_ESTABLISHED, c)
# advertise malformed AS4_PATH
e1.add_route('10.7.0.17/32', attribute='0x11 0x60 0x11223344')
@staticmethod
def check(env):
g1 = env.g1
e1 = env.e1
e2 = env.e2
def f():
for line in e1.log().split('\n'):
if 'UPDATE message error / Attribute Flags Error / 0x60110411223344' in line:
return True
return False
wait_for(f)
# check e2 is still established
g1.wait_for(BGP_FSM_ESTABLISHED, e2)
@staticmethod
def executor(env):
lookup_scenario("MalformedAs4Path").boot(env)
lookup_scenario("MalformedAs4Path").setup(env)
lookup_scenario("MalformedAs4Path").check(env)
@register_scenario
class MalformedNexthop(object):
"""
No.5 malformaed nexthop
"""
@staticmethod
def boot(env):
lookup_scenario("MalformedMpReachNlri").boot(env)
@staticmethod
def setup(env):
g1 = env.g1
e1 = env.e1
e2 = env.e2
for c in [e1, e2]:
g1.wait_for(BGP_FSM_ESTABLISHED, c)
# advertise malformed NEXT_HOP
# 0x0e: MP_REACH_NLRI
# 0x60: Optional, Transitive
# 0x01: AFI(IPv4)
# 0x01: SAFI(unicast)
# 0x10: Length of Next Hop Address
# 0xffffff00: Network address of Next Hop
# 0x00: Reserved
e1.add_route('10.7.0.17/32', attribute='0x0e 0x60 0x010110ffffff0000')
@staticmethod
def check(env):
g1 = env.g1
e1 = env.e1
e2 = env.e2
def f():
for line in e1.log().split('\n'):
if 'UPDATE message error / Attribute Flags Error / 0x600E08010110FFFFFF0000' in line:
return True
return False
wait_for(f)
# check e2 is still established
g1.wait_for(BGP_FSM_ESTABLISHED, e2)
@staticmethod
def executor(env):
lookup_scenario("MalformedNexthop").boot(env)
lookup_scenario("MalformedNexthop").setup(env)
lookup_scenario("MalformedNexthop").check(env)
@register_scenario
class MalformedRouteFamily(object):
"""
No.6 malformaed route family
"""
@staticmethod
def boot(env):
lookup_scenario("MalformedMpReachNlri").boot(env)
@staticmethod
def setup(env):
g1 = env.g1
e1 = env.e1
e2 = env.e2
for c in [e1, e2]:
g1.wait_for(BGP_FSM_ESTABLISHED, c)
# advertise malformed ROUTE_FAMILY
# 0x0e: MP_REACH_NLRI
# 0x60: Optional, Transitive
# 0x01: AFI(IPv4)
# 0x01: SAFI(unicast)
# 0x10: Length of Next Hop Address
# 0xffffff00: Network address of Next Hop
# 0x00: Reserved
e1.add_route('10.7.0.17/32', attribute='0x0e 0x60 0x0002011020010db800000000000000000000000100')
@staticmethod
def check(env):
g1 = env.g1
e1 = env.e1
e2 = env.e2
def f():
for line in e1.log().split('\n'):
if 'UPDATE message error / Attribute Flags Error / 0x600E150002011020010DB800000000000000000000000100' in line:
return True
return False
wait_for(f)
# check e2 is still established
g1.wait_for(BGP_FSM_ESTABLISHED, e2)
@staticmethod
def executor(env):
lookup_scenario("MalformedRouteFamily").boot(env)
lookup_scenario("MalformedRouteFamily").setup(env)
lookup_scenario("MalformedRouteFamily").check(env)
@register_scenario
class MalformedAsPathSegmentLengthInvalid(object):
"""
No.7 malformaed aspath segment length invalid
"""
@staticmethod
def boot(env):
lookup_scenario("MalformedMpReachNlri").boot(env)
@staticmethod
def setup(env):
g1 = env.g1
e1 = env.e1
e2 = env.e2
for c in [e1, e2]:
g1.wait_for(BGP_FSM_ESTABLISHED, c)
# advertise malformed AS_PATH SEGMENT LENGTH
# Send the attribute to the length and number of aspath is inconsistent
# Attribute Type 0x02 (AS_PATH)
# Attribute Flag 0x40 (well-known transitive)
# Attribute Value 0x02020000ffdc (
# segment type = 02
# segment length = 02 -> # correct value = 01
# as number = 65500 )
e1.add_route('10.7.0.17/32', attribute='0x02 0x40 0x0202ffdc')
@staticmethod
def check(env):
g1 = env.g1
e1 = env.e1
e2 = env.e2
def f():
for line in e1.log().split('\n'):
if 'UPDATE message error / Malformed AS_PATH / 0x4002040202FFDC' in line:
return True
return False
wait_for(f)
# check e2 is still established
g1.wait_for(BGP_FSM_ESTABLISHED, e2)
@staticmethod
def executor(env):
lookup_scenario("MalformedAsPathSegmentLengthInvalid").boot(env)
lookup_scenario("MalformedAsPathSegmentLengthInvalid").setup(env)
lookup_scenario("MalformedAsPathSegmentLengthInvalid").check(env)
@register_scenario
class MalformedNexthopLoopbackAddr(object):
"""
No.8 malformaed nexthop loopback addr
"""
@staticmethod
def boot(env):
lookup_scenario("MalformedMpReachNlri").boot(env)
@staticmethod
def setup(env):
g1 = env.g1
e1 = env.e1
e2 = env.e2
for c in [e1, e2]:
g1.wait_for(BGP_FSM_ESTABLISHED, c)
# Malformed Invalid NEXT_HOP Attribute
# Send the attribute of invalid nexthop
# next-hop 127.0.0.1 -> # correct value = other than loopback and 0.0.0.0 address
e1.add_route('10.7.0.17/32', nexthop='127.0.0.1')
@staticmethod
def check(env):
g1 = env.g1
e1 = env.e1
e2 = env.e2
def f():
for line in e1.log().split('\n'):
if 'UPDATE message error / Invalid NEXT_HOP Attribute / 0x4003047F000001' in line:
return True
return False
wait_for(f)
# check e2 is still established
g1.wait_for(BGP_FSM_ESTABLISHED, e2)
@staticmethod
def executor(env):
lookup_scenario("MalformedNexthopLoopbackAddr").boot(env)
lookup_scenario("MalformedNexthopLoopbackAddr").setup(env)
lookup_scenario("MalformedNexthopLoopbackAddr").check(env)
@register_scenario
class MalformedOriginType(object):
"""
No.9 malformaed origin type
"""
@staticmethod
def boot(env):
lookup_scenario("MalformedMpReachNlri").boot(env)
@staticmethod
def setup(env):
g1 = env.g1
e1 = env.e1
e2 = env.e2
for c in [e1, e2]:
g1.wait_for(BGP_FSM_ESTABLISHED, c)
# Invalid ORIGIN Attribute
# Send the attribute of origin type 4
# Attribute Type 0x01 (Origin)
# Attribute Flag 0x40 (well-known transitive)
# Attribute Value 0x04 (
# origin type = 04 -> # correct value = 01 or 02 or 03 )
e1.add_route('10.7.0.17/32', attribute='0x1 0x40 0x04')
@staticmethod
def check(env):
g1 = env.g1
e1 = env.e1
e2 = env.e2
def f():
for line in e1.log().split('\n'):
if 'UPDATE message error / Invalid ORIGIN Attribute / 0x40010104' in line:
return True
return False
wait_for(f)
# check e2 is still established
g1.wait_for(BGP_FSM_ESTABLISHED, e2)
@staticmethod
def executor(env):
lookup_scenario("MalformedOriginType").boot(env)
lookup_scenario("MalformedOriginType").setup(env)
lookup_scenario("MalformedOriginType").check(env)
class TestGoBGPBase(unittest.TestCase):
wait_per_retry = 5
retry_limit = 10
@classmethod
def setUpClass(cls):
idx = parser_option.test_index
base.TEST_PREFIX = parser_option.test_prefix
cls.parser_option = parser_option
cls.executors = []
if idx == 0:
print('unset test-index. run all test sequential')
for _, v in list(_SCENARIOS.items()):
for k, m in inspect.getmembers(v, inspect.isfunction):
if k == 'executor':
cls.executor = m
cls.executors.append(cls.executor)
elif idx not in _SCENARIOS:
print('invalid test-index. # of scenarios: {0}'.format(len(_SCENARIOS)))
sys.exit(1)
else:
for k, m in inspect.getmembers(_SCENARIOS[idx], inspect.isfunction):
if k == 'executor':
cls.executor = m
cls.executors.append(cls.executor)
def test(self):
for e in self.executors:
yield e
if __name__ == '__main__':
output = local("which docker 2>&1 > /dev/null ; echo $?", capture=True)
if int(output) != 0:
print("docker not found")
sys.exit(1)
nose.main(argv=sys.argv, addplugins=[OptionParser()],
defaultTest=sys.argv[0])
View Git Blame Copy Permalink