# Copyright (c) 2014 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

from autotest_lib.client.common_lib import error
from autotest_lib.client.cros import dhcp_handling_rule
from autotest_lib.client.cros import dhcp_packet
from autotest_lib.client.cros import dhcp_test_base

# Length of time the lease from the DHCP server is valid.
LEASE_TIME_SECONDS = 3600
# We'll fill in the subnet and give this address to the client.
INTENDED_IP_SUFFIX = "0.0.0.101"

class network_DhcpNak(dhcp_test_base.DhcpTestBase):
    """
    Tests a DHCP client's handling of NAK messages.

    Negotiates a lease, and then tests how the DHCP client processes
    NAKs in two scenarios. In the first scenario, the DHCP client
    is started anew, but with the cached lease on disk. In the
    second scenario, an already-running DHCP client is asked to
    renew its lease.

    In both scenarios, the NAK messages omit the DHCP server-id
    option. This is to emulate some DHCP servers (e.g. OpenBSD 4.6),
    which omit the DHCP server-id in NAK messages.
    """

    def common_setup(self):
        """
        Run common setup steps.
        """
        subnet_mask = self.ethernet_pair.interface_subnet_mask
        self.intended_ip = dhcp_test_base.DhcpTestBase.rewrite_ip_suffix(
            subnet_mask, self.server_ip, INTENDED_IP_SUFFIX)
        self.interface_name = self.ethernet_pair.peer_interface_name
        self.dhcp_options = {
            dhcp_packet.OPTION_SERVER_ID : self.server_ip,
            dhcp_packet.OPTION_SUBNET_MASK : subnet_mask,
            dhcp_packet.OPTION_IP_LEASE_TIME : LEASE_TIME_SECONDS,
            dhcp_packet.OPTION_REQUESTED_IP : self.intended_ip,
            dhcp_packet.OPTION_DNS_SERVERS : [],
            dhcp_packet.OPTION_DOMAIN_NAME : '',
            dhcp_packet.OPTION_DNS_DOMAIN_SEARCH_LIST : [],
            }
        self.negotiate_and_check_lease(self.dhcp_options)

    def reconnect_service(self):
        """
        Disconnect and reconnect Ethernet.

        Ask shill to disconnect and reconnect the Service for our
        virtual Ethernet link. This causes shill to shut down and
        restart dhcpcd for the link.
        """
        service = self.find_ethernet_service(self.interface_name)
        service.Disconnect()
        rules = [
            # Respond to DISCOVERY, but then NAK the REQUEST.
            dhcp_handling_rule.DhcpHandlingRule_RespondToDiscovery(
                self.intended_ip, self.server_ip, self.dhcp_options, {}),
            dhcp_handling_rule.DhcpHandlingRule_RejectRequest(),

            # Allow a successful negotiation the second time around.
            dhcp_handling_rule.DhcpHandlingRule_RespondToDiscovery(
                self.intended_ip, self.server_ip, self.dhcp_options, {}),
            dhcp_handling_rule.DhcpHandlingRule_RespondToRequest(
                self.intended_ip, self.server_ip, self.dhcp_options, {}),
            ]
        rules[-1].is_final_handler = True
        self.server.start_test(
            rules, dhcp_test_base.DHCP_NEGOTIATION_TIMEOUT_SECONDS)
        service.Connect()

    def force_dhcp_renew(self):
        """
        Force a DHCP renewal.

        Ask shill to renew the DHCP lease associated with our
        virtual Ethernet link.
        """
        rules = [
            # Reject REQUEST from renewal attempt.
            dhcp_handling_rule.DhcpHandlingRule_RejectRequest(),

            # Allow a successful negotiation after that.
            dhcp_handling_rule.DhcpHandlingRule_RespondToDiscovery(
                self.intended_ip, self.server_ip, self.dhcp_options, {}),
            dhcp_handling_rule.DhcpHandlingRule_RespondToRequest(
                self.intended_ip, self.server_ip, self.dhcp_options, {}),
            ]
        rules[-1].is_final_handler = True
        self.server.start_test(
            rules, dhcp_test_base.DHCP_NEGOTIATION_TIMEOUT_SECONDS)
        self.get_device(self.interface_name).RenewDHCPLease()

    def send_ack_then_nak(self):
        """
        Send an ACK followed by a NAK on re-connect to Ethernet.

        Ask shill to disconnect and reconnect the Service for our
        virtual Ethernet link. This causes shill to shut down and
        restart dhcpcd for the link.  Then perform a test where
        the server responds to a REQUEST with an ACK followed by
        an ACK.
        """
        service = self.find_ethernet_service(self.interface_name)
        service.Disconnect()
        rules = [
            # Respond to DISCOVERY, but then both ACK and NAK the REQUEST.
            dhcp_handling_rule.DhcpHandlingRule_RespondToDiscovery(
                self.intended_ip, self.server_ip, self.dhcp_options, {}),
            dhcp_handling_rule.DhcpHandlingRule_RejectAndRespondToRequest(
                self.intended_ip, self.server_ip, self.dhcp_options, {},
                False),
            ]
        rules[-1].is_final_handler = True
        self.server.start_test(
            rules, dhcp_test_base.DHCP_NEGOTIATION_TIMEOUT_SECONDS)
        service.Connect()

    def send_nak_then_ack_with_conflict(self):
        """
        Send an NAK followed by an ACK on re-connect to with address conflict.

        Ask shill to disconnect and reconnect the Service for our
        virtual Ethernet link. This causes shill to shut down and
        restart dhcpcd for the link.

        On reconnect, perform a test where the server responds to a
        REQUEST with a NAK followed by an ACK, however with a lease
        for an invalid address (the same IP address as the DHCP server).

        Ensure that the client rejects the invalid lease with a DECLINE,
        and that it also ignores the first OFFER for the same invalid
        address.
        """
        service = self.find_ethernet_service(self.interface_name)
        service.Disconnect()
        rules = [
            # Respond to DISCOVERY, but then both NAK then ACK the REQUEST,
            # supplying the server's own IP address.
            dhcp_handling_rule.DhcpHandlingRule_RespondToDiscovery(
                self.server_ip, self.server_ip, self.dhcp_options, {}),
            dhcp_handling_rule.DhcpHandlingRule_RejectAndRespondToRequest(
                self.server_ip, self.server_ip, self.dhcp_options, {},
                True),

            # The client should eventually reject this lease since this
            # address is in use.
            dhcp_handling_rule.DhcpHandlingRule_AcceptDecline(
                self.server_ip, self.dhcp_options, {}),

            # Offer up the same (invalid) IP address.
            dhcp_handling_rule.DhcpHandlingRule_RespondToDiscovery(
                self.server_ip, self.server_ip, self.dhcp_options, {}),

            # The client should ignore the previous offer and perform
            # another DISCOVER request.
            dhcp_handling_rule.DhcpHandlingRule_RespondToDiscovery(
                self.intended_ip, self.server_ip, self.dhcp_options, {}),
            dhcp_handling_rule.DhcpHandlingRule_RespondToRequest(
                self.intended_ip, self.server_ip, self.dhcp_options, {}),
            ]
        rules[-1].is_final_handler = True
        self.server.start_test(
            rules, dhcp_test_base.DHCP_NEGOTIATION_TIMEOUT_SECONDS)
        service.Connect()

    def send_nak_then_ack_then_verify(self):
        """
        Send an NAK followed by an ACK then verify client IP address.

        Ask shill to disconnect and reconnect the Service for our
        virtual Ethernet link. This causes shill to shut down and
        restart dhcpcd for the link.

        On reconnect, perform a test where the server responds to a
        REQUEST with a NAK followed by an ACK.  This method asserts
        that the client does not DECLINE this address.
        """
        service = self.find_ethernet_service(self.interface_name)
        service.Disconnect()

        # This rule serves two purposes: First it asserts that the client
        # does not send a DECLINE response.  Second, it waits until the
        # test timeout, by which time client will have completed an "ARP
        # self" operation to validate the offered IP adddres.
        decline_rule = dhcp_handling_rule.DhcpHandlingRule_AcceptDecline(
            self.intended_ip, self.dhcp_options, {})

        rules = [
            # Respond to DISCOVERY, but then both NAK then ACK the REQUEST,
            # supplying the server's own IP address.
            dhcp_handling_rule.DhcpHandlingRule_RespondToDiscovery(
                self.intended_ip, self.server_ip, self.dhcp_options, {}),
            dhcp_handling_rule.DhcpHandlingRule_RejectAndRespondToRequest(
                self.intended_ip, self.server_ip, self.dhcp_options, {},
                True),
            decline_rule
            ]
        rules[-1].is_final_handler = True
        self.server.start_test(
            rules, dhcp_test_base.DHCP_NEGOTIATION_TIMEOUT_SECONDS)
        service.Connect()
        self.server.wait_for_test_to_finish()

        # This is a negative test, since we expect the last rule to fail.
        if self.server.last_test_passed:
            raise error.TestFail('DHCP DECLINE message was received')
        elif self.server.current_rule != decline_rule:
            raise error.TestFail('Failed on %s rule' % self.server.current_rule)

        dhcp_config = self.get_interface_ipconfig(
                self.ethernet_pair.peer_interface_name)
        if dhcp_config is None:
            raise error.TestFail('Did not get a DHCP config')
        if dhcp_config[dhcp_test_base.DHCPCD_KEY_ADDRESS] != self.intended_ip:
            raise error.TestFail('Client did not attain expected address %s' %
                                 self.intended_ip)

    def test_body(self):
        """
        Entry point for this test.

        This is called from DhcpTestBase.run_once().
        """
        self.common_setup()
        for sub_test in (self.reconnect_service,
                         self.force_dhcp_renew,
                         self.send_ack_then_nak,
                         self.send_nak_then_ack_with_conflict):
            sub_test()
            self.server.wait_for_test_to_finish()
            if not self.server.last_test_passed:
                raise error.TestFail('Test failed (%s): active rule is %s' % (
                        sub_test.__name__, self.server.current_rule))

        # This method is outside the loop above since it performs its own
        # special verification.
        self.send_nak_then_ack_then_verify()