# Integration tests for the ldap server, using raw socket IO
#
# Tests for handling of malformed or large packets.
#
# Copyright (C) Catalyst.Net Ltd 2020
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#

import socket
import sys

import samba.tests
from samba.tests import TestCase


#
# LDAP Operations
#
SEARCH = b'\x63'

EQUALS = b'\xa3'


#
# ASN.1 Element types
#
BOOLEAN = b'\x01'
INTEGER = b'\x02'
OCTET_STRING = b'\x04'
NULL = b'\x05'
ENUMERATED = b'\x0a'
SEQUENCE = b'\x30'
SET = b'\x31'


#
# ASN.1 Helper functions.
#
def encode_element(ber_type, data):
    ''' Encode an ASN.1 BER element. '''
    if data is None:
        return ber_type + encode_length(0)
    return ber_type + encode_length(len(data)) + data


def encode_length(length):
    ''' Encode the length of an ASN.1 BER element.  '''

    if length > 0xFFFFFF:
        return b'\x84' + length.to_bytes(4, "big")
    if length > 0xFFFF:
        return b'\x83' + length.to_bytes(3, "big")
    if length > 0xFF:
        return b'\x82' + length.to_bytes(2, "big")
    if length > 0x7F:
        return b'\x81' + length.to_bytes(1, "big")
    return length.to_bytes(1, "big")


def encode_string(string):
    ''' Encode an octet string '''
    return encode_element(OCTET_STRING, string)


def encode_boolean(boolean):
    ''' Encode a boolean value '''
    if boolean:
        return encode_element(BOOLEAN, b'\xFF')
    return encode_element(BOOLEAN, b'\x00')


def encode_integer(integer):
    ''' Encode an integer value '''
    bit_len = integer.bit_length()
    byte_len = (bit_len // 8) + 1
    return encode_element(INTEGER, integer.to_bytes(byte_len, "big"))


def encode_enumerated(enum):
    ''' Encode an enumerated value '''
    return encode_element(ENUMERATED, enum.to_bytes(1, "big"))


def encode_sequence(sequence):
    ''' Encode a sequence '''
    return encode_element(SEQUENCE, sequence)


class RawLdapTest(TestCase):
    """A raw Ldap Test case."""

    def setUp(self):
        super(RawLdapTest, self).setUp()

        self.host = samba.tests.env_get_var_value('SERVER')
        self.port = 389
        self.socket = None
        self.connect()

    def tearDown(self):
        self.disconnect()
        super(RawLdapTest, self).tearDown()

    def disconnect(self):
        ''' Disconnect from and clean up the connection to the server '''
        if self.socket is None:
            return
        self.socket.close()
        self.socket = None

    def connect(self):
        ''' Open a socket stream connection to the server '''
        try:
            self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.socket.settimeout(10)
            self.socket.connect((self.host, self.port))
        except socket.error:
            self.socket.close()
            raise

    def send(self, req):
        ''' Send the request to the server '''
        try:
            self.socket.sendall(req)
        except socket.error:
            self.disconnect()
            raise

    def recv(self, num_recv=0xffff, timeout=None):
        ''' recv an array of bytes from the server '''
        data = None
        try:
            if timeout is not None:
                self.socket.settimeout(timeout)
            data = self.socket.recv(num_recv, 0)
            self.socket.settimeout(10)
            if len(data) == 0:
                self.disconnect()
                return None
        except socket.timeout:
            # We ignore timeout's as the ldap server will drop the connection
            # on the errors we're testing. So returning None on a timeout is
            # the desired behaviour.
            self.socket.settimeout(10)
        except socket.error:
            self.disconnect()
            raise
        return data

    def test_search_equals_maximum_permitted_size(self):
        '''
        Check that an LDAP search request equal to the maximum size is accepted
        '''
        #
        # Test is not compatable with python 2 so skip it for the
        # backports.
        #
        if sys.version_info < (3, 0):
            self.skipTest("Test is not python2 compatable")
            return

        # Lets build an ldap search packet to query the RootDSE
        header = encode_string(None)        # Base DN, ""
        header += encode_enumerated(0)      # Enumeration scope
        header += encode_enumerated(0)      # Enumeration dereference
        header += encode_integer(0)         # Integer size limit
        header += encode_integer(0)         # Integer time limit
        header += encode_boolean(False)     # Boolean attributes only

        #
        # build an equality search of the form x...x=y...y
        # With the length of x...x and y...y chosen to generate an
        # ldap request of 256000 bytes.
        x = encode_string(b'x' * 127974)
        y = encode_string(b'y' * 127979)
        equals = encode_element(EQUALS, x + y)
        trailer = encode_sequence(None)
        search = encode_element(SEARCH, header + equals + trailer)

        msg_no = encode_integer(1)
        packet = encode_sequence(msg_no + search)
        #
        # The length of the packet should be equal to the
        # Maximum length of a search query
        self.assertEqual(256000, len(packet))

        self.send(packet)
        data = self.recv()
        self.assertIsNotNone(data)

        # Should be a sequence
        self.assertEqual(SEQUENCE, data[0:1])

    def test_search_exceeds_maximum_permitted_size(self):
        '''
        Test that a search query longer than the maximum permitted
        size is rejected.
        '''
        #
        # Test is not compatable with python 2 so skip it for the
        # backports.
        #
        if sys.version_info < (3, 0):
            self.skipTest("Test is not python2 compatable")
            return

        # Lets build an ldap search packet to query the RootDSE
        header = encode_string(None)        # Base DN, ""
        header += encode_enumerated(0)      # Enumeration scope
        header += encode_enumerated(0)      # Enumeration dereference
        header += encode_integer(0)         # Integer size limit
        header += encode_integer(0)         # Integer time limit
        header += encode_boolean(False)     # Boolean attributes only

        #
        # build an equality search of the form x...x=y...y
        # With the length of x...x and y...y chosen to generate an
        # ldap request of 256001 bytes.
        x = encode_string(b'x' * 127979)
        y = encode_string(b'y' * 127975)
        equals = encode_element(EQUALS, x + y)
        trailer = encode_sequence(None)
        search = encode_element(SEARCH, header + equals + trailer)

        msg_no = encode_integer(1)
        packet = encode_sequence(msg_no + search)
        #
        # The length of the sequence data should be one greater than the
        # Maximum length of a search query
        self.assertEqual(256001, len(packet))

        self.send(packet)
        data = self.recv()
        #
        # The connection should be closed by the server and we should not
        # see any data.
        self.assertIsNone(data)
