#!/usr/bin/python2.6

#

# CDDL HEADER START

#

# The contents of this file are subject to the terms of the

# Common Development and Distribution License (the "License").

# You may not use this file except in compliance with the License.

#

# You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE

# or http://www.opensolaris.org/os/licensing.

# See the License for the specific language governing permissions

# and limitations under the License.

#

# When distributing Covered Code, include this CDDL HEADER in each

# file and include the License file at usr/src/OPENSOLARIS.LICENSE.

# If applicable, add the following below this CDDL HEADER, with the

# fields enclosed by brackets "[]" replaced with your own identifying

# information: Portions Copyright [yyyy] [name of copyright owner]

#

# CDDL HEADER END

#

#

# Copyright (c) 2012, Oracle and/or its affiliates. All rights reserved.

#

import rad.client as rad

import rad.util as ru

import common.tests as tests

import pwd

import os

import shutil

import socket

import ssl

import subprocess

# some useful constants

TEST_DIRNAME = "/tmp/rad_connect_test"

REMOTE_TEST_DIRNAME = TEST_DIRNAME + "_remote"

CERT_FILENAME = "cert.pem"

KEY_FILENAME = "key.pem"

AFUNIX_FILENAME = "unix"

# read commonly used environment variables

LOCAL_PORT_TCP = os.environ["LOCAL_PORT_TCP"]

LOCAL_PORT_TLS = os.environ["LOCAL_PORT_TLS"]

REMOTE_HOST_IPV4 = os.environ["REMOTE_HOST_IPV4"]

REMOTE_HOST_IPV6 = os.environ["REMOTE_HOST_IPV6"]

REMOTE_PORT_TCP = int(os.environ["REMOTE_PORT_TCP"])

REMOTE_PORT_TLS = int(os.environ["REMOTE_PORT_TCP"])

RUN_REMOTE_IPV4_TESTS = os.environ["REMOTE_TEST_IPV4"] == "yes"

RUN_REMOTE_IPV6_TESTS = os.environ["REMOTE_TEST_IPV6"] == "yes"

#

# default implementation of RADCommon.get_modules() for all classes using the

# TCP transport.

#

class TCP(object):

    def get_modules(self):

#

# default implementation of RADCommon.get_modules() for all classes using the

# TLS transport.

#

class TLS(object):

    def get_modules(self):

#

# Base class for all connection tests.

#

class ConnectTest(tests.RADTestObject):

    def get_keys(self):

	return [("type", "Authentication")]

    def get_name(self):

	domain = "com.oracle.solaris.rad.pam"

	return rad.Name(domain, self.get_keys())

    #

    # given a RadConnection object test its validity by retrieving the

    # Authentication object and reading the Connection Timeout value. This

    # value should be greater than zero. The default value is 180 seconds.

    #

    def verify_connection(self, rad_conn):

	""" verify that both the server and client report the same

	    connected user."""

	auth = rad_conn.get_object(self.get_name())

	co = auth.connectionTimeout

	# co should be greater than zero, the default value is 180

	if (co <= 0):

	    self.fail("unable to connect to RAD")

    #

    # Create a socket connection to the given host and port. This function

    # supports both IPv4 and IPv6 socket connection. Hosts must be either

    # a) IP addresses or b) resolvable to the appropriate type (v4/v6) of

    # addess. When in doubt, go with (a).

    #

    def _create_socket(self, host, port):

	addrinfo = socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM)

	family = addrinfo[0][0]

	address = addrinfo[0][4]

	s = socket.socket(family, socket.SOCK_STREAM)

	s.connect(address)

	return s

    #

    # A private function to create a TCP connection.

    #

    def _connect_tcp(self, host, port):

	return self._create_socket(host, port)

    #

    # A private function to create a TLS connection.

    #

    def _connect_tls(self, host, port, keyfile, certfile):

	s = self._connect_tcp(host, port)

	return ssl.wrap_socket(s, keyfile, certfile)

    #

    # A public function to create a TCP connection.

    #

    def connect_tcp(self, host, port, locale = None):

	s = self._connect_tcp(host, port)

	return rad.RadConnection(s, locale = locale)

    #

    # A public function to create a TLS connection.

    #

    def connect_tls(self, host, port, keyfile, certfile, locale = None):

	s = self._connect_tls(host, port, keyfile, certfile)

	return rad.RadConnection(s, locale = locale)

#

# Base class for all local connection tests.

#

class ConnectTestLocal(ConnectTest):

    #

    # override setUp.

    #

    def setUp(self):

	self.create_test_dir()

	super(ConnectTestLocal, self).setUp()

    #

    # override tearDown

    #

    def tearDown(self):

	super(ConnectTestLocal, self).tearDown()

	self.delete_test_dir()

    #

    # prepare the temporary test directory.

    #

    def create_test_dir(self):

	if not os.path.exists(TEST_DIRNAME):

	    os.makedirs(TEST_DIRNAME)

    #

    # cleanup the temporary test directory.

    #

    def delete_test_dir(self):

	if os.path.exists(TEST_DIRNAME):

	    shutil.rmtree(TEST_DIRNAME)

#

# Base class for all local AF_UNIX connection tests.

#

class ConnectTestLocalAFUNIX(ConnectTestLocal):

    def get_modules(self):

    def get_unix_path(self):

	return os.path.join(TEST_DIRNAME, AFUNIX_FILENAME)

    # This function verifies the connection and retrieves the logged used

    def get_server_username(self):

	# verify that RAD is up and running

	self.verify_connection(self._rc)

	# connect to RAD using UDS

	rad_conn = ru.connect_unix(self.get_unix_path())

	self.verify_connection(rad_conn)

	# retrieve and return the server username

	auth = rad_conn.get_object(self.get_name())

	svr_user = auth.user

	rad_conn.close()

	return svr_user

#

# Test RAD on the local machine using an AF_UNIX socket and peercred=true,

# the default.

#

class ConnectTestLocalAFUNIXauth(ConnectTestLocalAFUNIX):

    def get_auxargs(self):

	transport = "unix:path=" + self.get_unix_path()

	return ["-t", transport]

    def connectLocalAFUNIXauth(self):

	clnt_user = pwd.getpwuid(os.getuid()).pw_name

	svr_user = self.get_server_username()

	if clnt_user != svr_user:

	    self.fail("peercred failed. The client user should " \

		"be the same as the server user.")

    def test_connectLocalAFUNIXauth(self):

	""" connect to a local rad daemon using an af_unix socket """

	self.connectLocalAFUNIXauth()

#

# Test RAD on the local machine using an AF_UNIX socket and peercred=false

#

class ConnectTestLocalAFUNIXunauth(ConnectTestLocalAFUNIX):

    def get_auxargs(self):

	transport = "unix:path=" + self.get_unix_path() +\

	    ",peercred=false"

	return ["-t", transport]

    def connectLocalAFUNIXunauth(self):

	clnt_user = pwd.getpwuid(os.getuid()).pw_name

	svr_user = self.get_server_username()

	if clnt_user == svr_user:

	    self.fail("peercred failed. The client user should " \

		"not be the same as the server user.")

    def test_connectLocalAFUNIXunauth(self):

	""" connect to a local rad daemon using an af_unix socket and

	    peercred=false """

	self.connectLocalAFUNIXunauth()

#

# Test RAD on the local machine using the TCP transport.

#

class ConnectTestLocalTCP(TCP, ConnectTestLocal):

    def get_auxargs(self):

	transport = "tcp:localonly=false,port=" + str(LOCAL_PORT_TCP)

	return ["-t", transport]

    def connectLocalTCP(self, host):

	# verify that RAD is up and running

	self.verify_connection(self._rc)

	rad_conn = self.connect_tcp(host, LOCAL_PORT_TCP)

	self.verify_connection(rad_conn)

	rad_conn.close()

    def test_connectLocalTCP4(self):

	""" Connect to a local RAD daemon using IPv4 loopback """

	self.connectLocalTCP("127.0.0.1")

    def test_connectLocalTCP6(self):

	""" Connect to a local RAD daemon using IPv6 loopback """

	self.connectLocalTCP("::1")

#

# Test RAD on the local machine using the TLS transport.

#

class ConnectTestLocalTLS(TLS, ConnectTestLocal):

    def get_auxargs(self):

	certfile = os.path.join(TEST_DIRNAME, CERT_FILENAME)

	keyfile = os.path.join(TEST_DIRNAME, KEY_FILENAME)

	transport = "tls:localonly=false,port=" + \

	    str(LOCAL_PORT_TLS) + \

	    ",certificate=" + certfile + \

	    ",privatekey=" + keyfile + \

	    ",generate=true"

	return ["-t", transport]

    def connectLocalTLS(self, host):

	# verify that RAD is up and running.

	self.verify_connection(self._rc)

	certfile = os.path.join(TEST_DIRNAME, CERT_FILENAME)

	keyfile = os.path.join(TEST_DIRNAME, KEY_FILENAME)

	rad_conn = self.connect_tls(host, LOCAL_PORT_TLS,

	    keyfile, certfile)

	self.verify_connection(rad_conn)

	rad_conn.close()

    def test_connectLocalTLS4(self):

	""" Connect to a local RAD daemon using TLS over IPV4 loopback

	"""

	self.connectLocalTLS("127.0.0.1")

    def test_connectLocalTLS6(self):

	""" Connect to a local RAD daemon using TLS over IPV6 loopback

	"""

	self.connectLocalTLS("::1")

#

# Base class for all remote connection tests.

#

class ConnectTestRemote(ConnectTest):

    #

    # override the setUp method.

    #

    def setUp(self):

	if RUN_REMOTE_IPV4_TESTS or RUN_REMOTE_IPV6_TESTS:

	    # ssh to a remote machine and start a private copy of

	    # RAD.

	    cmd = ru.build_rad_cmd(modules = self.get_modules(),

		debug = False, root = REMOTE_TEST_DIRNAME,

		auxargs = self.get_auxargs())

	    host = "%s@%s" % (pwd.getpwuid(os.getuid()).pw_name,

		REMOTE_HOST_IPV4)

	    args = ["/usr/bin/ssh", "-q", "-e", "none","-o",

		"BatchMode yes", REMOTE_HOST_IPV4]

	    args.extend(cmd)

	    p = subprocess.Popen(args, 0, None,

		subprocess.PIPE, subprocess.PIPE)

	    rad_conn = rad.RadConnection(ru.ProcessPseudoSocket(p), None)

	    self._rc = rad_conn

	    name = self.get_name()

	    if name is not None:

		self._obj = self._rc.get_object(name)

#

# Test a remote RAD using the TCP transport.

#

class ConnectTestRemoteTCP(TCP, ConnectTestRemote):

    def get_auxargs(self):

	transport = "tcp:localonly=false,port=" + str(REMOTE_PORT_TCP)

	return ["-t", transport]

    #

    # Connect to a remote RAD using TCP.

    #

    def connectRemoteTCP(self, host):

	# verify that RAD is up and running

	self.verify_connection(self._rc)

	rad_conn = self.connect_tcp(host, REMOTE_PORT_TCP)

	self.verify_connection(rad_conn)

	rad_conn.close()

    def test_connectRemoteTCP4(self):

	""" Connect to a remote RAD daemon using IPv4 """

	if RUN_REMOTE_IPV4_TESTS:

	    self.connectRemoteTCP(REMOTE_HOST_IPV4)

    def test_connectRemoteTCP6(self):

	""" Connect to a remote RAD daemon using IPv6 """

	if RUN_REMOTE_IPV6_TESTS:

	    self.connectRemoteTCP(REMOTE_HOST_IPV6)

#

# Test a remote RAD using the TLS transport.

#

class ConnectTestRemoteTLS(TLS, ConnectTestRemote):

    def get_auxargs(self):

	certfile = os.path.join(REMOTE_TEST_DIRNAME, CERT_FILENAME)

	keyfile = os.path.join(REMOTE_TEST_DIRNAME, KEY_FILENAME)

	transport = "tls:localonly=false,port=" + \

	    str(REMOTE_PORT_TLS) + \

	    ",certificate=" + certfile + \

	    ",privatekey=" + keyfile

	return ["-t", transport]

    #

    # Connect to a remote RAD using the TLS transport.

    #

    def connectRemoteTLS(self, host):

	certfile = os.path.join(REMOTE_TEST_DIRNAME, CERT_FILENAME)

	keyfile = os.path.join(REMOTE_TEST_DIRNAME, KEY_FILENAME)

	# verify that RAD is up and running

	self.verify_connection(self._rc)

	rad_conn = self.connect_tls(host, REMOTE_PORT_TLS,

	    keyfile, certfile)

	self.verify_connection(rad_conn)

	rad_conn.close()

    def test_connectRemoteTLS4(self):

	""" Connect to a remote RAD daemon using IPv4 """

	if RUN_REMOTE_IPV4_TESTS:

	    self.connectRemoteTLS(REMOTE_HOST_IPV4)

    def test_connectRemoteTLS6(self):

	""" Connect to a remote RAD daemon using IPv6 """

	if RUN_REMOTE_IPV6_TESTS:

	    self.connectRemoteTLS(REMOTE_HOST_IPV6)

#

# Test a remote RAD started with localonly=true using the TCP transport.

#

class ConnectTestRemoteTCPlocalonly(TCP, ConnectTestRemote):

    def get_auxargs(self):

	transport = "tcp:localonly=true,port=" + str(REMOTE_PORT_TCP)

	return ["-t", transport]

    #

    # Connect to a remote RAD, started with localonly=true, using TCP

    #

    def connectRemoteTCPlocalonly(self, host):

	# verify that RAD is up and running

	self.verify_connection(self._rc)

	# verify that the server does not accept remote connections

	try:

	    conn = self.connect_tcp(host, REMOTE_PORT_TCP)

	    self.verify_connection(conn)

	    # We should not get here

	    self.fail("Connection to a remote server started " \

		"with 'localonly=true' erroneously successful.")

	except IOError:

	    # Expected behavior.

	    pass

    def test_connectRemoteTCP4localonly(self):

	""" Connect to a remote RAD daemon using IPv4 with

	localonly=true """

	if RUN_REMOTE_IPV4_TESTS:

	    self.connectRemoteTCPlocalonly(REMOTE_HOST_IPV4)

    def test_connectRemoteTCP6localonly(self):

	""" Connect to a rmeote RAD daemon using IPV6 with

	localonly=true """

	if RUN_REMOTE_IPV6_TESTS:

	    self.connectRemoteTCPlocalonly(REMOTE_HOST_IPV6)

#

# Test a remote RAD started with localonly=true using the TLS transport.

#

class ConnectTestRemoteTLSlocalonly(TLS, ConnectTestRemote):

    def get_auxargs(self):

	certfile = os.path.join(REMOTE_TEST_DIRNAME, CERT_FILENAME)

	keyfile = os.path.join(REMOTE_TEST_DIRNAME, KEY_FILENAME)

	transport = "tls:localonly=true,port=" + \

	    str(REMOTE_PORT_TLS) + \

	    ",certificate=" + certfile + \

	    ",privatekey=" + keyfile

	return ["-t", transport]

    #

    # Connect to a remote RAD, started with localonly=true, using TLS

    #

    def connectRemoteTLSlocalonly(self, host):

	certfile = os.path.join(REMOTE_TEST_DIRNAME, CERT_FILENAME)

	keyfile = os.path.join(REMOTE_TEST_DIRNAME, KEY_FILENAME)

	# verify that RAD is up and running

	self.verify_connection(self._rc)

	# verify that the server does not accept remote connections

	try:

	    conn = self.connect_tls(host, REMOTE_PORT_TLS,

		certfile, keyfile)

	    self.fail("Connection to a remote server started " \

		"with localonly=true erroneously successful.")

	except IOError:

	    # Expected behavior.

	    pass

    def test_connectRemoteTLS4localonly(self):

	""" Connect to a remote RAD daemon using TLS over IPv4 with

	localonly=true. """

	if RUN_REMOTE_IPV4_TESTS:

	    self.connectRemoteTLSlocalonly(REMOTE_HOST_IPV4)

    def test_connectRemoteTLS6localonly(self):

	""" Connect to a remote RAD daemon using TLS over IPv6 with

	localonly=true. """

	if RUN_REMOTE_IPV6_TESTS:

	    self.connectRemoteTLSlocalonly(REMOTE_HOST_IPV6)

#

# Test a remote RAD using the TLS transport and a bad certificate.

#

class ConnectTestRemoteTLSbadcert(TLS, ConnectTestRemote):

    def get_auxargs(self):

	certfile = os.path.join(REMOTE_TEST_DIRNAME, CERT_FILENAME)

	keyfile = os.path.join(REMOTE_TEST_DIRNAME, KEY_FILENAME)

	transport = "tls:localonly=false,port=" + \

	    str(REMOTE_PORT_TLS) + \