#!/usr/bin/env python3
#
#  Copyright (c) 2019, The OpenThread Authors.
#  All rights reserved.
#
#  Redistribution and use in source and binary forms, with or without
#  modification, are permitted provided that the following conditions are met:
#  1. Redistributions of source code must retain the above copyright
#     notice, this list of conditions and the following disclaimer.
#  2. Redistributions in binary form must reproduce the above copyright
#     notice, this list of conditions and the following disclaimer in the
#     documentation and/or other materials provided with the distribution.
#  3. Neither the name of the copyright holder nor the
#     names of its contributors may be used to endorse or promote products
#     derived from this software without specific prior written permission.
#
#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
#  ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
#  LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
#  CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
#  SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
#  INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
#  CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
#  ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
#  POSSIBILITY OF SUCH DAMAGE.
#
# This is a test script for checking layer fields against a given test.pcap.
#

import logging
import unittest

from pktverify import layer_fields
from pktverify.addrs import EthAddr, ExtAddr, Ipv6Addr
from pktverify.bytes import Bytes
from pktverify.consts import REAL_LAYER_NAMES, VALID_LAYER_NAMES
from pktverify.layer_fields_container import LayerFieldsContainer
from pktverify.null_field import nullField
from pktverify.packet import Packet
from pktverify.pcap_reader import PcapReader


class TestLayerFields(unittest.TestCase):

    def test(self):
        logging.basicConfig(level=logging.DEBUG)
        pkts = PcapReader.read("test.pcap")

        for i, p in enumerate(pkts):

            logging.info("check packet #%d", i + 1)

            for layer_name in VALID_LAYER_NAMES:
                if layer_name == 'lowpan':  # we already checked 6lowpan
                    continue

                layer = getattr(p, layer_name)
                if hasattr(p._packet, layer_name):
                    if layer_name in REAL_LAYER_NAMES:
                        self.assertTrue(layer)
                    checker = getattr(self, '_test_' + layer_name, None)
                    if checker is None:
                        continue

                    try:
                        checker(p)
                    except Exception:
                        layer.show()
                        raise
                else:
                    if layer_name in REAL_LAYER_NAMES:
                        self.assertFalse(layer)

            for layer in p._packet.layers:
                self._check_missing_fields(p, layer.layer_name, layer)

    def _test_coap(self, p):
        coap = p.coap
        self.assertIsInstance(coap.version, int)
        self.assertIsInstance(coap.type, int)
        self.assertIsInstance(coap.token_len, int)
        self.assertIsInstance(coap.code, int)
        self.assertIsInstance(coap.mid, int)
        self.assertIsInstanceOrNull(coap.token, int)
        self.assertIsInstanceOrNull(coap.opt.uri_path_recon, str)
        self.assertIsInstanceOrNull(coap.payload, bytearray)

        print(p.coap.tlv.type, p.coap.tlv)
        assert isinstance(coap.tlv, LayerFieldsContainer), repr(coap.tlv)
        self.assertIsInstanceOrNull(coap.tlv.type, list)

    def _test_mle(self, p):
        mle = p.mle

        self._must_have_wpan_aux_sec(p)

        self.assertIsInstance(mle.cmd, int)
        self.assertIsInstanceOrNull(mle.tlv.mode.receiver_on_idle, int)
        self.assertIsInstanceOrNull(mle.tlv.mode.reserved1, int)
        self.assertIsInstanceOrNull(mle.tlv.mode.reserved2, int)
        self.assertIsInstanceOrNull(mle.tlv.mode.device_type_bit, int)
        self.assertIsInstanceOrNull(mle.tlv.mode.network_data, int)
        self.assertIsInstanceOrNull(mle.tlv.challenge, Bytes)
        self.assertIsInstanceOrNull(mle.tlv.scan_mask.r, int)
        self.assertIsInstanceOrNull(mle.tlv.scan_mask.e, int)
        self.assertIsInstanceOrNull(mle.tlv.version, int)
        self.assertIsInstanceOrNull(mle.tlv.source_addr, int)
        self.assertIsInstanceOrNull(mle.tlv.active_tstamp, int)
        self.assertIsInstanceOrNull(mle.tlv.leader_data.partition_id, int)
        self.assertIsInstanceOrNull(mle.tlv.leader_data.weighting, int)
        self.assertIsInstanceOrNull(mle.tlv.leader_data.data_version, int)
        self.assertIsInstanceOrNull(mle.tlv.leader_data.stable_data_version, int)
        self.assertIsInstanceOrNull(mle.tlv.leader_data.router_id, int)

    def _test_wpan(self, p):
        wpan = p.wpan
        self.assertIsInstance(wpan.fcf, int)
        self.assertIsInstance(wpan.fcs, int)
        self.assertIsInstance(wpan.security, int)
        self.assertIsInstance(wpan.pending, int)
        self.assertIsInstance(wpan.ack_request, int)
        self.assertIsInstance(wpan.pan_id_compression, int)
        self.assertIsInstance(wpan.seqno_suppression, int)
        self.assertIsInstance(wpan.ie_present, int)
        self.assertIsInstance(wpan.dst_addr_mode, int)
        self.assertIsInstance(wpan.version, int)
        self.assertIsInstance(wpan.src_addr_mode, int)

        self.assertIsInstance(wpan.seq_no, int)

        if not wpan.is_ack:
            self.assertIsInstanceOrNull(wpan.dst_pan, int)
            self.assertIsInstanceOrNull(wpan.dst16, int)
            self.assertIsInstanceOrNull(wpan.src16, int)
            self.assertIsInstanceOrNull(wpan.src64, ExtAddr)
            self.assertIsInstanceOrNull(wpan.dst64, ExtAddr)

        if wpan.aux_sec:
            self._must_have_wpan_aux_sec(p)

    def _must_have_wpan_aux_sec(self, p):
        wpan = p.wpan
        self.assertIsInstanceOrNull(wpan.aux_sec.sec_suite, int)
        self.assertIsInstanceOrNull(wpan.aux_sec.security_control_field, int)
        self.assertIsInstanceOrNull(wpan.aux_sec.sec_level, int)
        self.assertIsInstanceOrNull(wpan.aux_sec.key_id_mode, int)
        self.assertIsInstanceOrNull(wpan.aux_sec.frame_counter_suppression, int)
        self.assertIsInstanceOrNull(wpan.aux_sec.asn_in_nonce, int)
        self.assertIsInstanceOrNull(wpan.aux_sec.reserved, int)
        self.assertIsInstanceOrNull(wpan.aux_sec.frame_counter, int)
        self.assertIsInstanceOrNull(wpan.aux_sec.key_source, int)
        self.assertIsInstanceOrNull(wpan.aux_sec.key_index, int)

    def assertIsInstanceOrNull(self, field, type):
        if field is not nullField:
            self.assertIsInstance(field, type)

    def _test_thread_bl(self, p):
        thread_bl = p.thread_bl
        self.assertTrue(thread_bl)

        self.assertIsInstanceOrNull(thread_bl.tlv.target_eid, Ipv6Addr)
        self.assertIsInstanceOrNull(thread_bl.tlv.ml_eid, ExtAddr)
        self.assertIsInstanceOrNull(thread_bl.tlv.last_transaction_time, int)
        self.assertIsInstanceOrNull(p.thread_meshcop.tlv.net_name, list)

    def _test_thread_meshcop(self, p: Packet):
        thread_meshcop = p.thread_meshcop

        for layer in sorted(p.layers, key=lambda l: l.layer_name):
            if 'thread_meshcop.tlv.commissioner_sess_id' in layer._layer._all_fields:
                self.assertIsInstance(thread_meshcop.tlv.commissioner_sess_id, int)

            if 'thread_meshcop.tlv.net_name' in layer._layer._all_fields:
                self.assertIsInstance(thread_meshcop.tlv.net_name, list)

            if 'thread_meshcop.tlv.channel_page' in layer._layer._all_fields:
                self.assertIsInstance(thread_meshcop.tlv.channel_page, int)

            if 'thread_meshcop.tlv.channel' in layer._layer._all_fields:
                self.assertIsInstance(thread_meshcop.tlv.channel, list)

            if 'thread_meshcop.tlv.chan_mask_page' in layer._layer._all_fields:
                self.assertIsInstance(thread_meshcop.tlv.chan_mask_page, int)
            if 'thread_meshcop.tlv.chan_mask_len' in layer._layer._all_fields:
                self.assertIsInstance(thread_meshcop.tlv.chan_mask_len, int)
            if 'thread_meshcop.tlv.chan_mask_mask' in layer._layer._all_fields:
                self.assertIsInstance(thread_meshcop.tlv.chan_mask_mask, int)

            if 'thread_meshcop.tlv.panid' in layer._layer._all_fields:
                self.assertIsInstance(thread_meshcop.tlv.panid, int)

            if 'thread_meshcop.tlv.ml_prefix' in layer._layer._all_fields:
                self.assertIsInstance(thread_meshcop.tlv.ml_prefix, Bytes)

            if 'thread_meshcop.tlv.master_key' in layer._layer._all_fields:
                self.assertIsInstance(thread_meshcop.tlv.master_key, Bytes)

            if 'thread_meshcop.tlv.pskc' in layer._layer._all_fields:
                self.assertIsInstance(thread_meshcop.tlv.pskc, Bytes)

            if 'thread_meshcop.tlv.sec_policy_rot' in layer._layer._all_fields:
                self.assertIsInstance(thread_meshcop.tlv.sec_policy_rot, int)

            if 'thread_meshcop.tlv.sec_policy_o' in layer._layer._all_fields:
                self.assertIsInstance(thread_meshcop.tlv.sec_policy_o, int)

            if 'thread_meshcop.tlv.sec_policy_n' in layer._layer._all_fields:
                self.assertIsInstance(thread_meshcop.tlv.sec_policy_n, int)

            if 'thread_meshcop.tlv.sec_policy_r' in layer._layer._all_fields:
                self.assertIsInstance(thread_meshcop.tlv.sec_policy_r, int)

            if 'thread_meshcop.tlv.sec_policy_c' in layer._layer._all_fields:
                self.assertIsInstance(thread_meshcop.tlv.sec_policy_c, int)

            if 'thread_meshcop.tlv.sec_policy_b' in layer._layer._all_fields:
                self.assertIsInstance(thread_meshcop.tlv.sec_policy_b, int)

            if 'thread_meshcop.tlv.pan_id' in layer._layer._all_fields:
                self.assertIsInstance(thread_meshcop.tlv.pan_id, list)

            if 'thread_meshcop.tlv.xpan_id' in layer._layer._all_fields:
                self.assertIsInstance(thread_meshcop.tlv.xpan_id, Bytes)

            for field in layer._layer._all_fields:
                if field.startswith('thread_meshcop') and not layer_fields.is_layer_field(field) and field not in (
                        'thread_meshcop.tlv', 'thread_meshcop.tlv.type', 'thread_meshcop.tlv.len8'):
                    print('found %s = %s in layer %s' % (
                        field,
                        layer._layer.get_field(field),
                        layer.layer_name,
                    ))

    def _test_icmpv6(self, p):
        icmpv6 = p.icmpv6
        self.assertTrue(p.icmpv6)

        self.assertIsInstance(icmpv6.type, int)
        self.assertIsInstance(icmpv6.code, int)
        self.assertIsInstance(icmpv6.checksum, int)
        self.assertIsInstanceOrNull(icmpv6.reserved, int)
        self.assertIsInstanceOrNull(icmpv6.nd.na.flag.s, int)
        self.assertIsInstanceOrNull(icmpv6.nd.na.flag.o, int)
        self.assertIsInstanceOrNull(icmpv6.nd.na.flag.r, int)
        self.assertIsInstanceOrNull(icmpv6.nd.na.flag.rsv, int)
        self.assertIsInstanceOrNull(icmpv6.nd.ra.cur_hop_limit, int)
        self.assertIsInstanceOrNull(icmpv6.mldr.nb_mcast_records, int)
        self.assertIsInstanceOrNull(icmpv6.nd.ns.target_address, Ipv6Addr)
        self.assertIsInstanceOrNull(icmpv6.mldr.mar.multicast_address, list)

    def get_field(self, p: Packet, f):
        secs = f.split('.')
        assert len(secs) >= 2
        v = p
        for sec in secs:
            v = getattr(v, sec)

        return v

    def _test_6lowpan(self, p):
        lowpan = p.lowpan
        assert lowpan is getattr(p, '6lowpan')
        self.assertIsInstanceOrNull(lowpan.src, Ipv6Addr)
        self.assertIsInstanceOrNull(lowpan.dst, Ipv6Addr)
        self.assertIsInstanceOrNull(lowpan.udp.src, int)
        self.assertIsInstanceOrNull(lowpan.udp.dst, int)
        self.assertIsInstanceOrNull(lowpan.udp.checksum, int)
        self.assertIsInstanceOrNull(lowpan.frag.size, int)
        self.assertIsInstanceOrNull(lowpan.frag.tag, int)
        self.assertIsInstanceOrNull(lowpan.frag.offset, int)
        self.assertIsInstanceOrNull(lowpan.nhc.pattern, list)
        self.assertIsInstanceOrNull(lowpan.nhc.udp.checksum, int)
        self.assertIsInstanceOrNull(lowpan.nhc.udp.ports, int)

        self.assertIsInstanceOrNull(lowpan.pattern, list)
        self.assertIsInstanceOrNull(lowpan.iphc.tf, int)
        self.assertIsInstanceOrNull(lowpan.iphc.nh, int)
        self.assertIsInstanceOrNull(lowpan.iphc.hlim, int)
        self.assertIsInstanceOrNull(lowpan.iphc.cid, int)
        self.assertIsInstanceOrNull(lowpan.iphc.sac, int)
        self.assertIsInstanceOrNull(lowpan.iphc.sam, int)
        self.assertIsInstanceOrNull(lowpan.iphc.m, int)
        self.assertIsInstanceOrNull(lowpan.iphc.dac, int)
        self.assertIsInstanceOrNull(lowpan.iphc.dam, int)
        self.assertIsInstanceOrNull(lowpan.iphc.sctx.prefix, Bytes)
        self.assertIsInstanceOrNull(lowpan.iphc.dctx.prefix, Bytes)

    def _test_ip(self, p):
        pass

    def _test_ipv6(self, p):
        pass

    def _test_udp(self, p):
        pass

    def _test_eth(self, p):
        eth = p.eth
        self.assertIsInstance(eth.src, EthAddr)
        self.assertIsInstance(eth.dst, EthAddr)
        self.assertIsInstance(eth.type, int)

    def _check_missing_fields(self, p, layer_name, _layer):
        for f in sorted(_layer._all_fields.keys(), reverse=True):
            if f.startswith('_ws') or f.startswith('data'):
                continue

            logging.info('_check_missing_fields in layer %s: %s = %r' % (layer_name, f, _layer._all_fields[f]))
            if f in {
                    '', 'icmpv6.checksum.status', 'ip.ttl.lncb', 'wpan.aux_sec.key_source.bytes', 'wpan.src64.origin'
            }:
                # TODO: handle these fields
                continue

            v = _layer._all_fields[f]

            if layer_fields.is_layer_field_container(f):
                continue

            try:
                rv = self.get_field(p, f)
                self.assertIsNot(rv, nullField)

                parser = layer_fields._LAYER_FIELDS[f]
                if isinstance(parser, layer_fields._first):
                    parser = parser._sub_parse

                if parser in (layer_fields._raw_hex, layer_fields._hex, layer_fields._raw_hex_rev, layer_fields._dec,
                              layer_fields._auto):
                    self.assertIsInstance(rv, int)
                elif isinstance(parser, layer_fields._list):
                    self.assertIsInstance(rv, list)
                elif parser is layer_fields._ipv6_addr:
                    self.assertIsInstance(rv, Ipv6Addr)
                elif parser is layer_fields._eth_addr:
                    self.assertIsInstance(rv, EthAddr)
                elif parser is layer_fields._ext_addr:
                    self.assertIsInstance(rv, ExtAddr)
                elif parser is layer_fields._str:
                    self.assertIsInstance(rv, str)
                elif parser is layer_fields._bytes:
                    self.assertIsInstance(rv, Bytes)
                elif parser is layer_fields._payload:
                    self.assertIsInstance(rv, bytearray)
                elif parser is layer_fields._float:
                    self.assertIsInstance(rv, float)
                else:
                    raise NotImplementedError(parser)
            except Exception:
                logging.info('checking [%s] %s=%r, %r, %r (%d)' %
                             (layer_name, f, v, v.get_default_value(), v.raw_value, len(v.fields)))
                raise