packet lib: ipv6: support options for Hop-by-Hop Options header and destination header

Signed-off-by: itoyuichi <ito.yuichi0@gmail.com>
Signed-off-by: FUJITA Tomonori <fujita.tomonori@lab.ntt.co.jp>

2 files changed, 131 insertions, 0 deletions

diff --git a/ryu/lib/packet/ipv6.py b/ryu/lib/packet/ipv6.py
index 6bcc6086..060e678c 100644
--- a/ryu/lib/packet/ipv6.py
+++ b/ryu/lib/packet/ipv6.py
@@ -168,3 +168,62 @@ class header(stringify.StringifyMixin):
         pass
 
 # TODO: implement a class for routing header
+
+
+class option(stringify.StringifyMixin):
+    """IPv6 (RFC 2460) Options header encoder/decoder class.
+
+    This is used with ryu.lib.packet.ipv6.hop_opts or
+                      ryu.lib.packet.ipv6.dst_opts.
+
+    An instance has the following attributes at least.
+    Most of them are same to the on-wire counterparts but in host byte order.
+    __init__ takes the corresponding args in this order.
+
+    .. tabularcolumns:: |l|L|
+
+    ============== =======================================
+    Attribute      Description
+    ============== =======================================
+    type\_         option type.
+    len\_          the length of data. -1 if type\_ is 0.
+    data           an option value. None if len\_ is 0 or -1.
+    ============== =======================================
+    """
+
+    _PACK_STR = '!BB'
+    _MIN_LEN = struct.calcsize(_PACK_STR)
+
+    def __init__(self, type_, len_, data):
+        self.type_ = type_
+        self.len_ = len_
+        self.data = data
+
+    @classmethod
+    def parser(cls, buf):
+        (type_, ) = struct.unpack_from('!B', buf)
+        if not type_:
+            cls_ = cls(type_, -1, None)
+        else:
+            data = None
+            (type_, len_) = struct.unpack_from(cls._PACK_STR, buf)
+            if len_:
+                form = "%ds" % len_
+                (data, ) = struct.unpack_from(form, buf, cls._MIN_LEN)
+            cls_ = cls(type_, len_, data)
+        return cls_
+
+    def serialize(self):
+        data = None
+        if not self.type_:
+            data = struct.pack('!B', self.type_)
+        elif not self.len_:
+            data = struct.pack(self._PACK_STR, self.type_, self.len_)
+        else:
+            form = "%ds" % self.len_
+            data = struct.pack(self._PACK_STR + form, self.type_,
+                               self.len_, self.data)
+        return data
+
+    def __len__(self):
+        return self._MIN_LEN + self.len_
diff --git a/ryu/tests/unit/packet/test_ipv6.py b/ryu/tests/unit/packet/test_ipv6.py
index e4c0e5e1..22b6a52b 100644
--- a/ryu/tests/unit/packet/test_ipv6.py
+++ b/ryu/tests/unit/packet/test_ipv6.py
@@ -120,3 +120,75 @@ class Test_ipv6(unittest.TestCase):
 
     def test_len(self):
         eq_(len(self.ip), 40)
+
+
+class Test_option(unittest.TestCase):
+
+    def setUp(self):
+        self.type_ = 5
+        self.data = '\x00\x00'
+        self.len_ = len(self.data)
+        self.opt = ipv6.option(self.type_, self.len_, self.data)
+        self.form = '!BB%ds' % self.len_
+        self.buf = struct.pack(self.form, self.type_, self.len_, self.data)
+
+    def tearDown(self):
+        pass
+
+    def test_init(self):
+        eq_(self.type_, self.opt.type_)
+        eq_(self.len_, self.opt.len_)
+        eq_(self.data, self.opt.data)
+
+    def test_parser(self):
+        _res = ipv6.option.parser(self.buf)
+        if type(_res) is tuple:
+            res = _res[0]
+        else:
+            res = _res
+        eq_(self.type_, res.type_)
+        eq_(self.len_, res.len_)
+        eq_(self.data, res.data)
+
+    def test_serialize(self):
+        buf = self.opt.serialize()
+        res = struct.unpack_from(self.form, buf)
+        eq_(self.type_, res[0])
+        eq_(self.len_, res[1])
+        eq_(self.data, res[2])
+
+    def test_len(self):
+        eq_(len(self.opt), 2 + self.len_)
+
+
+class Test_option_pad1(Test_option):
+
+    def setUp(self):
+        self.type_ = 0
+        self.len_ = -1
+        self.data = None
+        self.opt = ipv6.option(self.type_, self.len_, self.data)
+        self.form = '!B'
+        self.buf = struct.pack(self.form, self.type_)
+
+    def test_serialize(self):
+        buf = self.opt.serialize()
+        res = struct.unpack_from(self.form, buf)
+        eq_(self.type_, res[0])
+
+
+class Test_option_padN(Test_option):
+
+    def setUp(self):
+        self.type_ = 1
+        self.len_ = 0
+        self.data = None
+        self.opt = ipv6.option(self.type_, self.len_, self.data)
+        self.form = '!BB'
+        self.buf = struct.pack(self.form, self.type_, self.len_)
+
+    def test_serialize(self):
+        buf = self.opt.serialize()
+        res = struct.unpack_from(self.form, buf)
+        eq_(self.type_, res[0])
+        eq_(self.len_, res[1])