# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license

import io
import unittest

import dns.rdata
import dns.rdtypes.svcbbase
import dns.rrset
from dns.tokenizer import Tokenizer

from tests.util import here

class SVCBTestCase(unittest.TestCase):
    def check_valid_inputs(self, inputs):
        expected = inputs[0]
        for text in inputs:
            rr = dns.rdata.from_text('IN', 'SVCB', text)
            new_text = rr.to_text()
            self.assertEqual(expected, new_text)

    def check_invalid_inputs(self, inputs):
        for text in inputs:
            with self.assertRaises((dns.exception.SyntaxError, ValueError)):
                dns.rdata.from_text('IN', 'SVCB', text)

    def test_svcb_general_invalid(self):
        invalid_inputs = (
            # Duplicate keys
            "1 . alpn=h2 alpn=h3",
            "1 . alpn=h2 key1=h3",
            # Quoted keys
            "1 . \"alpn=h2\"",
            # Invalid space
            "1 . alpn= h2",
            "1 . alpn =h2",
            "1 . alpn = h2",
            "1 . alpn= \"h2\"",
            "1 . =alpn",
        )
        self.check_invalid_inputs(invalid_inputs)

    def test_svcb_mandatory(self):
        valid_inputs = (
            "1 . mandatory=\"alpn,no-default-alpn\" alpn=\"h2\" no-default-alpn",
            "1 . mandatory=alpn,no-default-alpn alpn=h2 no-default-alpn",
            "1 . mandatory=key1,key2 alpn=h2 no-default-alpn",
            "1 . mandatory=alpn,no-default-alpn key1=\\002h2 key2",
            "1 . key0=\\000\\001\\000\\002 alpn=h2 no-default-alpn",
            "1 . alpn=h2 no-default-alpn mandatory=alpn,no-default-alpn",
        )
        self.check_valid_inputs(valid_inputs)

        invalid_inputs = (
            # empty
            "1 . mandatory=",
            "1 . mandatory",
            # unknown key
            "1 . mandatory=foo",
            # key 0
            "1 . mandatory=key0",
            "1 . mandatory=key0,alpn",
            # missing key
            "1 . mandatory=alpn",
            # duplicate
            "1 . mandatory=alpn,alpn alpn=h2",
            # invalid escaping
            "1 . mandatory=\\alpn alpn=h2",
            # empty wire format
            "1 . key0",
            "1 . key0=",
            # 0 in wire format
            "1 . key0=\\000\\000",
            # invalid length in wire format
            "1 . key0=\\000",
            # out of order in wire format
            "1 . key0=\\000\\002\\000\\001 alpn=h2 no-default-alpn",
            # leading zeros
            "1 . mandatory=key1,key002 alpn=h2 no-default-alpn",
        )
        self.check_invalid_inputs(invalid_inputs)

    def test_svcb_alpn(self):
        valid_inputs_two_items = (
            "1 . alpn=\"h2,h3\"",
            "1 . alpn=h2,h3",
            "1 . alpn=h\\050,h3",
            "1 . alpn=\"h\\050,h3\"",
            "1 . alpn=\\h2,h3",
            "1 . alpn=\"h2\\,h3\"",
            "1 . alpn=h2\\,h3",
            "1 . alpn=h2\\044h3",
            "1 . key1=\\002h2\\002h3",
        )
        self.check_valid_inputs(valid_inputs_two_items)

        valid_inputs_one_item = (
            "1 . alpn=\"h2\\\\,h3\"",
            "1 . alpn=h2\\\\,h3",
            "1 . alpn=h2\\092\\044h3",
            "1 . key1=\\005h2,h3",
        )
        self.check_valid_inputs(valid_inputs_one_item)

        invalid_inputs = (
            "1 . alpn",
            "1 . alpn=",
            "1 . alpn=h2,,h3",
            "1 . alpn=01234567890abcdef01234567890abcdef01234567890abcdef"
                     "01234567890abcdef01234567890abcdef01234567890abcdef"
                     "01234567890abcdef01234567890abcdef01234567890abcdef"
                     "01234567890abcdef01234567890abcdef01234567890abcdef"
                     "01234567890abcdef01234567890abcdef01234567890abcdef"
                     "01234567890abcdef",
            "1 . alpn=\",h2,h3\"",
            "1 . alpn=\"h2,h3,\"",
            "1 . key1",
            "1 . key1=",
            "1 . key1=\\000",
            "1 . key1=\\002x",
        )
        self.check_invalid_inputs(invalid_inputs)

    def test_svcb_no_default_alpn(self):
        valid_inputs = (
            "1 . alpn=\"h2\" no-default-alpn",
            "1 . alpn=\"h2\" no-default-alpn=\"\"",
            "1 . alpn=\"h2\" key2",
            "1 . alpn=\"h2\" key2=\"\"",
        )
        self.check_valid_inputs(valid_inputs)

        invalid_inputs = (
            "1 . no-default-alpn",
            "1 . no-default-alpn=\"\"",
            "1 . key2",
            "1 . key2=\"\"",
            "1 . alpn=h2 no-default-alpn=foo",
            "1 . alpn=h2 no-default-alpn=",
            "1 . alpn=h2 key2=foo",
            "1 . alpn=h2 key2=",
        )
        self.check_invalid_inputs(invalid_inputs)

    def test_svcb_port(self):
        valid_inputs = (
            "1 . port=\"53\"",
            "1 . port=53",
            "1 . key3=\\000\\053",
        )
        self.check_valid_inputs(valid_inputs)

        invalid_inputs = (
            "1 . port",
            "1 . port=",
            "1 . port=53x",
            "1 . port=x53",
            "1 . port=53,54",
            "1 . port=53\\,54",
            "1 . port=65536",
            "1 . key3",
            "1 . key3=",
            "1 . key3=\\000",
        )
        self.check_invalid_inputs(invalid_inputs)

    def test_svcb_ipv4hint(self):
        valid_inputs = (
            "1 . ipv4hint=\"0.0.0.0,1.1.1.1\"",
            "1 . ipv4hint=0.0.0.0,1.1.1.1",
            "1 . key4=\\000\\000\\000\\000\\001\\001\\001\\001",
        )
        self.check_valid_inputs(valid_inputs)

        invalid_inputs = (
            "1 . ipv4hint",
            "1 . ipv4hint=",
            "1 . ipv4hint=1234",
            "1 . ipv4hint=1\\.2.3.4",
            "1 . ipv4hint=1.2.3.4\\,2.3.4.5",
            "1 . key4=",
            "1 . key4=123",
        )
        self.check_invalid_inputs(invalid_inputs)

    def test_svcb_ech(self):
        valid_inputs = (
            "1 . ech=\"Zm9vMA==\"",
            "1 . ech=Zm9vMA==",
            "1 . key5=foo0",
            "1 . key5=\\102\\111\\111\\048",
        )
        self.check_valid_inputs(valid_inputs)

        invalid_inputs = (
            "1 . ech",
            "1 . ech=",
            "1 . ech=Zm9vMA",
            "1 . ech=\\090m9vMA==",
            "1 . key5",
            "1 . key5=",
        )
        self.check_invalid_inputs(invalid_inputs)

    def test_svcb_ipv6hint(self):
        valid_inputs = (
            "1 . ipv6hint=\"::4,1::\"",
            "1 . ipv6hint=::4,1::",
            "1 . key6=\\000\\000\\000\\000\\000\\000\\000\\000"
                     "\\000\\000\\000\\000\\000\\000\\000\\004"
                     "\\000\\001\\000\\000\\000\\000\\000\\000"
                     "\\000\\000\\000\\000\\000\\000\\000\\000",
        )
        self.check_valid_inputs(valid_inputs)

        invalid_inputs = (
            "1 . ipv6hint",
            "1 . ipv6hint=",
            "1 . ipv6hint=1234",
            "1 . ipv6hint=1\\::2",
            "1 . ipv6hint=::1\\,::2",
            "1 . ipv6hint",
            "1 . key6",
            "1 . key6=",
            "1 . key6=123",
        )
        self.check_invalid_inputs(invalid_inputs)

    def test_svcb_unknown(self):
        valid_inputs_one_key = (
            "1 . key23=\"key45\"",
            "1 . key23=key45",
            "1 . key23=key\\052\\053",
            "1 . key23=\"key\\052\\053\"",
            "1 . key23=\\107\\101\\121\\052\\053",
        )
        self.check_valid_inputs(valid_inputs_one_key)

        valid_inputs_one_key_empty = (
            "1 . key23",
            "1 . key23=\"\"",
        )
        self.check_valid_inputs(valid_inputs_one_key_empty)

        invalid_inputs_one_key = (
            "1 . key65536=foo",
            "1 . key24= key48",
        )
        self.check_invalid_inputs(invalid_inputs_one_key)

        valid_inputs_two_keys = (
            "1 . key24 key48",
            "1 . key24=\"\" key48",
        )
        self.check_valid_inputs(valid_inputs_two_keys)

    def test_svcb_wire(self):
        valid_inputs = (
            "1 . mandatory=\"alpn,port\" alpn=\"h2\" port=\"257\"",
            "\\# 24 0001 00 0000000400010003 00010003026832 000300020101",
        )
        self.check_valid_inputs(valid_inputs)

        everything = \
            "100 foo.com. mandatory=\"alpn,port\" alpn=\"h2,h3\" " \
            "             no-default-alpn port=\"12345\" ech=\"abcd\" " \
            "             ipv4hint=1.2.3.4,4.3.2.1 ipv6hint=1::2,3::4" \
            "             key12345=\"foo\""
        rr = dns.rdata.from_text('IN', 'SVCB', everything)
        rr2 = dns.rdata.from_text('IN', 'SVCB', rr.to_generic().to_text())
        self.assertEqual(rr, rr2)

        invalid_inputs = (
            # As above, but the keys are out of order.
            "\\# 24 0001 00 0000000400010003 000300020101 00010003026832",
            # As above, but the mandatory keys don't match
            "\\# 24 0001 00 0000000400010002 00010003026832 000300020101",
            "\\# 24 0001 00 0000000400010004 00010003026832 000300020101",
            # Alias form shouldn't have parameters.
            "\\# 08 0000 000300020101",
            # no-default-alpn requires alpn
            "\\# 07 0001 00 00020000",
        )
        self.check_invalid_inputs(invalid_inputs)

    def test_misc_escape(self):
        rdata = dns.rdata.from_text('in', 'svcb', '1 . alpn=\\010\\010')
        expected = '1 . alpn="\\\\010\\\\010"'
        self.assertEqual(rdata.to_text(), expected)
        with self.assertRaises(dns.exception.SyntaxError):
            dns.rdata.from_text('in', 'svcb', '1 . alpn=\\0')
        with self.assertRaises(dns.exception.SyntaxError):
            dns.rdata.from_text('in', 'svcb', '1 . alpn=\\00')
        with self.assertRaises(dns.exception.SyntaxError):
            dns.rdata.from_text('in', 'svcb', '1 . alpn=\\00q')
        with self.assertRaises(dns.exception.SyntaxError):
            dns.rdata.from_text('in', 'svcb', '1 . alpn=\\256')
        # This doesn't usually get exercised, so we do it directly.
        gp = dns.rdtypes.svcbbase.GenericParam.from_value('\\001\\002')
        expected = '"\\001\\002"'
        self.assertEqual(gp.to_text(), expected)

    def test_svcb_spec_test_vectors(self):
        text_file = here("svcb_test_vectors.text")
        text_tokenizer = Tokenizer(open(text_file), filename=text_file)
        generic_file = here("svcb_test_vectors.generic")
        generic_tokenizer = Tokenizer(open(generic_file), filename=generic_file)

        while True:
            while True:
                text_token = text_tokenizer.get()
                if text_token.is_eol():
                    continue
                break
            while True:
                generic_token = generic_tokenizer.get()
                if generic_token.is_eol():
                    continue
                break
            self.assertEqual(text_token.ttype, generic_token.ttype)
            if text_token.is_eof():
                break
            self.assertTrue(text_token.is_identifier)
            text_tokenizer.unget(text_token)
            generic_tokenizer.unget(generic_token)
            text_rdata = dns.rdata.from_text('IN', 'SVCB', text_tokenizer)
            generic_rdata = dns.rdata.from_text('IN', 'SVCB', generic_tokenizer)
            self.assertEqual(text_rdata, generic_rdata)

    def test_svcb_spec_failure_cases(self):
        failure_cases = (
            # This example has multiple instances of the same SvcParamKey
            "1 foo.example.com. key123=abc key123=def",
            # In the next examples the SvcParamKeys are missing their values.
            "1 foo.example.com. mandatory",
            "1 foo.example.com. alpn",
            "1 foo.example.com. port",
            "1 foo.example.com. ipv4hint",
            "1 foo.example.com. ipv6hint",
            # The "no-default-alpn" SvcParamKey value MUST be empty (Section 6.1).
            "1 foo.example.com. no-default-alpn=abc",
            # In this record a mandatory SvcParam is missing (Section 7).
            "1 foo.example.com. mandatory=key123",
            # The "mandatory" SvcParamKey MUST not be included in mandatory list
            # (Section 7).
            "1 foo.example.com. mandatory=mandatory",
            # Here there are multiple instances of the same SvcParamKey in the
            # mandatory list (Section 7).
            "1 foo.example.com. mandatory=key123,key123 key123=abc",
        )
        self.check_invalid_inputs(failure_cases);

    def test_alias_mode(self):
        rd = dns.rdata.from_text('in', 'svcb', '0 .')
        self.assertEqual(len(rd.params), 0)
        self.assertEqual(rd.target, dns.name.root)
        self.assertEqual(rd.to_text(), '0 .')
        rd = dns.rdata.from_text('in', 'svcb', '0 elsewhere.')
        self.assertEqual(rd.target, dns.name.from_text('elsewhere.'))
        self.assertEqual(len(rd.params), 0)
        # provoke 'parameters in AliasMode' from text.
        with self.assertRaises(dns.exception.SyntaxError):
            dns.rdata.from_text('in', 'svcb', '0 elsewhere. alpn=h2')
        # provoke 'parameters in AliasMode' from wire too.
        wire = bytes.fromhex('0000000000000400010003')
        with self.assertRaises(dns.exception.FormError):
            dns.rdata.from_wire('in', 'svcb', wire, 0, len(wire))

    def test_immutability(self):
        alpn = dns.rdtypes.svcbbase.ALPNParam.from_value(['h2', 'h3'])
        with self.assertRaises(TypeError):
            alpn.ids[0] = 'foo'
        with self.assertRaises(TypeError):
            del alpn.ids[0]
        with self.assertRaises(TypeError):
            alpn.ids = 'foo'
        with self.assertRaises(TypeError):
            del alpn.ids

    def test_alias_not_compressed(self):
        rrs = dns.rrset.from_text('elsewhere.', 300, 'in', 'svcb',
                                  '0 elseWhere.')
        output = io.BytesIO()
        compress = {}
        rrs.to_wire(output, compress)
        wire = output.getvalue()
        # Just one of these assertions is enough, but we do both to show
        # the bug we're checking is fixed.
        assert not wire.endswith(b'\xc0\x00')
        assert wire.endswith(b'\x09elseWhere\x00')