#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
# 
#   http://www.apache.org/licenses/LICENSE-2.0
# 
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#

import os, common
import subprocess
from proton import *
from common import Skipped


class SslTest(common.Test):

    def __init__(self, *args):
        common.Test.__init__(self, *args)

    def setup(self):
        try:
            self.t_server = Transport()
            self.server = SSL(self.t_server)
            self.server.init(SSL.MODE_SERVER)
            self.t_client = Transport()
            self.client = SSL(self.t_client)
            self.client.init(SSL.MODE_CLIENT)
        except SSLUnavailable, e:
            raise Skipped(e)

    def teardown(self):
        self.t_client = None
        self.t_server = None

    def _pump(self):
        while True:
            out_client = self.t_client.output(1024)
            out_server = self.t_server.output(1024)
            if out_client: self.t_server.input(out_client)
            if out_server: self.t_client.input(out_server)
            if not out_client and not out_server: break

    def _testpath(self, file):
        """ Set the full path to the certificate,keyfile, etc. for the test.
        """
        return os.path.join(os.path.dirname(__file__),
                            "ssl_db/%s" % file)

    def test_defaults(self):
        """ By default, both the server and the client support anonymous
        ciphers - they should connect without need for a certificate.
        """
        client_conn = Connection()
        self.t_client.bind(client_conn)
        server_conn = Connection()
        self.t_server.bind(server_conn)

        # check that no SSL connection exists
        assert not self.server.cipher_name()
        assert not self.client.protocol_name()

        client_conn.open()
        server_conn.open()
        self._pump()

        # now SSL should be active
        assert self.server.cipher_name() is not None
        assert self.client.protocol_name() is not None

        client_conn.close()
        server_conn.close()
        self._pump()


    def test_server_certificate(self):
        """ Test that anonymous clients can still connect to a server that has
        a certificate configured.
        """
        self.server.set_credentials(self._testpath("server-certificate.pem"),
                                    self._testpath("server-private-key.pem"),
                                    "server-password")
        client_conn = Connection()
        self.t_client.bind(client_conn)
        server_conn = Connection()
        self.t_server.bind(server_conn)
        client_conn.open()
        server_conn.open()
        self._pump()
        assert self.client.protocol_name() is not None
        client_conn.close()
        server_conn.close()
        self._pump()

    def test_server_authentication(self):
        """ Simple SSL connection with authentication of the server
        """
        self.server.set_credentials(self._testpath("server-certificate.pem"),
                                    self._testpath("server-private-key.pem"),
                                    "server-password")

        self.client.set_trusted_ca_db(self._testpath("ca-certificate.pem"))
        self.client.set_peer_authentication( SSL.VERIFY_PEER )

        client_conn = Connection()
        self.t_client.bind(client_conn)
        server_conn = Connection()
        self.t_server.bind(server_conn)
        client_conn.open()
        server_conn.open()
        self._pump()
        assert self.client.protocol_name() is not None
        client_conn.close()
        server_conn.close()
        self._pump()

    def test_client_authentication(self):
        """ Force the client to authenticate.
        """
        # note: when requesting client auth, the server _must_ send its
        # certificate, so make sure we configure one!
        self.server.set_credentials(self._testpath("server-certificate.pem"),
                                    self._testpath("server-private-key.pem"),
                                    "server-password")
        self.server.set_trusted_ca_db(self._testpath("ca-certificate.pem"))
        self.server.set_peer_authentication( SSL.VERIFY_PEER,
                                             self._testpath("ca-certificate.pem") )

        # give the client a certificate, but let's not require server authentication
        self.client.set_credentials(self._testpath("client-certificate.pem"),
                                    self._testpath("client-private-key.pem"),
                                    "client-password")
        self.client.set_peer_authentication( SSL.ANONYMOUS_PEER )

        client_conn = Connection()
        self.t_client.bind(client_conn)
        server_conn = Connection()
        self.t_server.bind(server_conn)
        client_conn.open()
        server_conn.open()
        self._pump()
        assert self.client.protocol_name() is not None
        client_conn.close()
        server_conn.close()
        self._pump()


    def test_client_server_authentication(self):
        """ Require both client and server to mutually identify themselves.
        """
        self.server.set_credentials(self._testpath("server-certificate.pem"),
                                    self._testpath("server-private-key.pem"),
                                    "server-password")
        self.server.set_trusted_ca_db(self._testpath("ca-certificate.pem"))
        self.server.set_peer_authentication( SSL.VERIFY_PEER,
                                             self._testpath("ca-certificate.pem") )

        self.client.set_credentials(self._testpath("client-certificate.pem"),
                                    self._testpath("client-private-key.pem"),
                                    "client-password")
        self.client.set_trusted_ca_db(self._testpath("ca-certificate.pem"))
        self.client.set_peer_authentication( SSL.VERIFY_PEER )

        client_conn = Connection()
        self.t_client.bind(client_conn)
        server_conn = Connection()
        self.t_server.bind(server_conn)
        client_conn.open()
        server_conn.open()
        self._pump()
        assert self.client.protocol_name() is not None
        client_conn.close()
        server_conn.close()
        self._pump()

    def test_server_only_authentication(self):
        """ Client verifies server, but server does not verify client.
        """
        self.server.set_credentials(self._testpath("server-certificate.pem"),
                                    self._testpath("server-private-key.pem"),
                                    "server-password")
        self.server.set_peer_authentication( SSL.ANONYMOUS_PEER )

        self.client.set_credentials(self._testpath("client-certificate.pem"),
                                    self._testpath("client-private-key.pem"),
                                    "client-password")
        self.client.set_trusted_ca_db(self._testpath("ca-certificate.pem"))
        self.client.set_peer_authentication( SSL.VERIFY_PEER )

        client_conn = Connection()
        self.t_client.bind(client_conn)
        server_conn = Connection()
        self.t_server.bind(server_conn)
        client_conn.open()
        server_conn.open()
        self._pump()
        assert self.client.protocol_name() is not None
        client_conn.close()
        server_conn.close()
        self._pump()