#!/usr/bin/python
#
# Copyright (C) 2014, 2015 Olaf Kirch <okir@suse.de>
#

import twopence
import susetest
import suselog
import re

journal = None
client = None
server = None

# twopence.setDebugLevel(0)

##################################################################
# Set up the test suite
##################################################################
def setup():
	global client, server, journal

	config = susetest.Config("tcpd")

	journal = config.journal

	journal.beginGroup("setup")

	client = config.target("client")
	client.defaultUser = "testuser"

	server = config.target("server")
	server.defaultUser = "testuser"

	if not client.ipaddr:
		journal.fatal("No IP address set for client");
	if not server.ipaddr:
		journal.fatal("No IP address set for server");


##################################################################
# Helper functions
##################################################################
def run_tcpdmatch(node, args):
	return node.runOrFail("/usr/sbin/tcpdmatch -d -i /dev/null " + args, user = 'root')

##################################################################
# Check the output of tcpdmatch
# As it checks the various rules, it will print any number of
# "denied" messages, and possibly one "granted" message.
# If any rule grants access, then we have a winner.
##################################################################
def tcpdmatch_verdict(status):
	lines = str(status.stdout).split('\n')
	for l in lines:
		if l.find("access:") >= 0:
			if l.find("granted") >= 0:
				return "granted"
			if l.find("denied") < 0:
				return "failed"
	return "denied"

def tcpdmatch_expect(node, service, address, expected_result):

	journal.beginTest("Matching service=%s address=%s, expect result=%s"
				% (service, address, expected_result))
	status = run_tcpdmatch(node, service + " " + address)
	if not status:
		return

	result = tcpdmatch_verdict(status)
	if result == expected_result:
		journal.success("Good, tcpdmatch returns expected result")
	else:
		journal.failure("Oops: tcpdmatch returns %s, but it should have returned %s" % (result, expected_result))


def tcpdmatch_test(node):

	hosts_allow = '''
rsyncd : localhost : ALLOW
rsyncd : 127.0.0.1 : ALLOW

one    : 192.168.9. : DENY
one    : 192.168. : ALLOW

two    : 192.168.9.0/24 : DENY
two    : 192.168.0.0/16 : ALLOW

three  : 192.168.9.0/255.255.255.0 : DENY
three  : 192.168.0.0/255.255.0.0 : ALLOW

four   : @MYHOSTNAME@ : ALLOW

six    : [3ffe:505:2:1::babe] : ALLOW
six    : [3ffe:505:3:1::/64] : ALLOW
six    : [3ffe:505:4:1::]/64 : ALLOW
six    : [3ffe:505:5:1::]/64 : DENY
'''

	hosts_deny = '''
sshd   : ALL : ALLOW
ALL    : ALL
'''

	journal.beginGroup("tcpdmatch")
	journal.beginTest("initialize tcpdmatch testing")
	status = node.run("hostname -f")
	if not status:
		journal.error("cannot get fully qualified hostname")
		return
	fqdn = str(status.stdout).strip()
	if not fqdn:
		journal.error("cannot get fully qualified hostname")
		return

	hosts_allow = re.sub('@MYHOSTNAME@', fqdn, hosts_allow)

	if not node.sendbuffer("hosts.allow", hosts_allow, user = 'root'):
		journal.error("Unable to send hosts.allow file")
		return
	if not node.sendbuffer("hosts.deny", hosts_deny, user = 'root'):
		journal.error("Unable to send hosts.deny file")
		return

	tcpdmatch_expect(node, "rsyncd",	"localhost",		"granted")
	tcpdmatch_expect(node, "rsyncd",	"127.0.0.1",		"granted")
	tcpdmatch_expect(node, "rsyncd",	"::1",			"granted")
	tcpdmatch_expect(node, "rsyncd",	"192.168.1.1",		"denied")

	tcpdmatch_expect(node, "one",		"192.168.1.1",		"granted")
	tcpdmatch_expect(node, "one",		"192.168.9.1",		"denied")

	tcpdmatch_expect(node, "two",		"192.168.1.1",		"granted")
	tcpdmatch_expect(node, "two",		"192.168.9.1",		"denied")

	tcpdmatch_expect(node, "three",		"192.168.1.1",		"granted")
	tcpdmatch_expect(node, "three",		"192.168.9.1",		"denied")

	# This test case is currently broken, because the
	# find_inet_addr() code used by tcpdmatch has a few
	# fundamental issues
	#tcpdmatch_expect(node, "four",		fqdn,			"granted")

	tcpdmatch_expect(node, "six",		"3ffe:505:2:1::babe",	"granted")
	tcpdmatch_expect(node, "six",		"3ffe:505:2:2::1",	"denied")
	tcpdmatch_expect(node, "six",		"3ffe:505:3:1::1",	"granted")
	tcpdmatch_expect(node, "six",		"3ffe:505:4:1::1",	"granted")
	tcpdmatch_expect(node, "six",		"3ffe:505:5:1::1",	"denied")

	# node.run("rm -f hosts.allow")

##################################################################
# Enable rsh
##################################################################
def rsh_enable():
	if server.run("test -d /usr/lib/systemd/system"):
		return rsh_enable_systemd()

	return rsh_enable_xinetd()

##################################################################
# Enable rsh in a systemd world
##################################################################
def rsh_enable_systemd():

	rsh_socket = '''
[Unit]
Description=BSD rsh service

[Socket]
ListenStream=514
Accept=yes

[Install]
WantedBy=sockets.target
'''

	rsh_at_service = '''
[Unit]
Description=BSD rshd
DefaultDependencies=no
Requires=rsh.socket
After=rsh.socket
After=local-fs.target

[Service]
Type=simple
ExecStart=-@/usr/sbin/tcpd /usr/sbin/in.rshd
StandardInput=socket

[Install]
WantedBy=multi-user.target
Also=rsh.socket
'''

	journal.beginTest("initialize tcpd rsh testing")

	if not server.sendbuffer("/usr/lib/systemd/system/rsh.socket", rsh_socket, user = 'root'):
		journal.error("Unable to install systemd unit file");
		return
	if not server.sendbuffer("/usr/lib/systemd/system/rsh@.service", rsh_at_service, user = 'root'):
		journal.error("Unable to install systemd unit file");
		return
	if not server.run("systemctl daemon-reload", user = "root") or \
	   not server.run("systemctl start rsh.socket", user = "root"):
		journal.error("Failed to start rsh.socket")
		return

##################################################################
# Enable rsh in an xinetd world
##################################################################
def xinetd_enable_service(node, service):

	journal.beginTest("%s: enable service %s in xinetd" % (node.name, service))
	filename = "/etc/xinetd.d/" + service

	data = node.recvbuffer(filename, user = 'root')
	if not data:
		journal.error("unable to download %s" % filename)
		return False
	
	result = []
	for line in str(data).split('\n'):
		words = line.split(" \t=")
		if words[0] == 'disable':
			line = "disable = no"
		result.append(line)
	data = '\n'.join(result) + '\n'
	if not node.sendbuffer(filename, data, user = 'root'):
		journal.error("failed to upload %s" % filename)
		return False

	# When xinetd detects that there are no services, it will
	# simply exit. So just a reload may fail.
	if not node.run("rcxinetd restart", user = 'root'):
		journal.error("unable to reload xinetd")
		return False

	journal.success()

def rsh_enable_xinetd():

	return xinetd_enable_service(server, "rsh")

##################################################################
# Test rsh with tcpwrappers
##################################################################
def test_rsh_acls():

	journal.beginGroup("rsh")
	rsh_enable()

	# Eric's test image creator is currently broken
	server.run("chown testuser ~testuser", user = 'root')

	# Install a valid rhosts file in testuser's account
	if not server.sendbuffer(".rhosts", "%s testuser\n" % client.ipaddr, user = 'testuser'):
		journal.error("Unable to install .rhosts file");
		return

	fqdn = "client.testing.suse.de"
	domain = "testing.suse.de"

	if not server.addHostEntry(client.ipaddr, fqdn):
		journal.error("Unable to update hosts file");
		return

	# By default, deny everything (except for sshd, which twopence may be relying on)
	if not rsh_send_host_deny("sshd : ALL : ALLOW", "ALL : ALL : DENY"):
		journal.error("Unable to send hosts.deny file")
		return

	journal.beginTest("test ALL:ALL:ALLOW")
	if not rsh_send_host_allow("ALL : ALL : ALLOW"):
		return
	rsh_should_succeed(server.ipaddr)

	journal.beginTest("test empty hosts.allow")
	if not rsh_send_host_allow():
		return
	rsh_should_fail(server.ipaddr)

	journal.beginTest("test rshd:ALL:ALLOW")
	if not rsh_send_host_allow("in.rshd : ALL : ALLOW"):
		return
	rsh_should_succeed(server.ipaddr)

	journal.beginTest("test rshd: <ipaddr> :ALLOW")
	if not rsh_send_host_allow("in.rshd : %s : ALLOW" % client.ipaddr):
		return
	rsh_should_succeed(server.ipaddr)

	journal.beginTest("test rshd:ALL EXCEPT <ipaddr>:ALLOW")
	if not rsh_send_host_allow("in.rshd : ALL EXCEPT %s: ALLOW" % client.ipaddr):
		return
	rsh_should_fail(server.ipaddr)

	journal.beginTest("test rshd:ALL EXCEPT <ipaddr>:DENY")
	if not rsh_send_host_allow("in.rshd : ALL EXCEPT %s: DENY" % client.ipaddr, "in.rshd : ALL : ALLOW"):
		return
	rsh_should_succeed(server.ipaddr)

	ipnetwork = re.sub("\.[0-9]*$", "", client.ipaddr)
	journal.beginTest("test rshd: <ipnetwork> :ALLOW")
	if not rsh_send_host_allow("in.rshd : %s. : ALLOW" % ipnetwork):
		return
	rsh_should_succeed(server.ipaddr)

	journal.beginTest("test rshd:ALL EXCEPT <ipnetwork>:DENY")
	if not rsh_send_host_allow("in.rshd : ALL EXCEPT %s.: DENY" % ipnetwork, "in.rshd : ALL : ALLOW"):
		return
	rsh_should_succeed(server.ipaddr)

	journal.beginTest("test rshd:ALL EXCEPT <ipaddr/netmask>:DENY")
	if not rsh_send_host_allow("in.rshd : %s.0/255.255.255.0 : ALLOW" % ipnetwork):
		return
	rsh_should_succeed(server.ipaddr)

	journal.beginTest("test rshd:ALL EXCEPT <ipaddr/prefix>:DENY")
	if not rsh_send_host_allow("in.rshd : %s.0/24 : ALLOW" % ipnetwork):
		return
	rsh_should_succeed(server.ipaddr)

	if not server.sendbuffer(".rhosts", "%s testuser\n" % fqdn, user = 'testuser'):
		journal.error("Unable to install .rhosts file");
		return

	journal.beginTest("test rshd: <hostname> :ALLOW")
	if not rsh_send_host_allow("in.rshd : %s : ALLOW" % fqdn):
		return
	rsh_should_succeed(server.ipaddr)

	journal.beginTest("test rshd:ALL EXCEPT <hostname>:DENY")
	if not rsh_send_host_allow("in.rshd : ALL EXCEPT %s: DENY" % fqdn, "in.rshd : ALL : ALLOW"):
		return
	rsh_should_succeed(server.ipaddr)

	journal.beginTest("test rshd: .<domainname> :ALLOW")
	if not rsh_send_host_allow("in.rshd : .%s : ALLOW" % domain):
		return
	rsh_should_succeed(server.ipaddr)


def rsh_send_host_deny(*lines):
	data = "\n".join(lines) + "\n"
	if not server.sendbuffer("/etc/hosts.deny", data, user = 'root'):
		journal.error("Unable to send hosts.allow file")
		return False
	return True

def rsh_send_host_allow(*lines):
	data = "\n".join(lines) + "\n"
	if not server.sendbuffer("/etc/hosts.allow", data, user = 'root'):
		journal.error("Unable to send hosts.allow file")
		return False
	return True


def rsh_should_succeed(hostname):
	global client

	status = client.run("rsh %s echo hello" % hostname, user = 'testuser')
	if not status:
		journal.failure("rsh failed (but was expected to succeed)")
		return False

	response = str(status.stdout).strip()
	if response != "hello":
		journal.failure("rsh produced unexpected output \"%s\"" % response);
		return False
	
	journal.success("rsh succeeded and returned string \"%s\"" % response)
	return True

def rsh_should_fail(hostname):
	global client

	status = client.run("rsh %s echo hello" % hostname, user = 'testuser')
	if not status:
		journal.success("rsh access was denied, as expected")
		return True
	
	journal.failure("rsh succeeded - access should have been denied")
	return True


def main():
	setup()
	tcpdmatch_test(server)
	test_rsh_acls()

	journal.writeReport()

main()
