mirror of
https://github.com/freedombox/FreedomBox.git
synced 2026-02-04 08:13:38 +00:00
524 lines
16 KiB
Python
524 lines
16 KiB
Python
"""
|
|
If I produce a listener that just echoes the parameters, I can validate the response:
|
|
|
|
import httplib, urllib
|
|
|
|
params = urllib.urlencode({'@number': 12524, '@type': 'issue', '@action': 'show'})
|
|
headers = {"Content-type": "application/x-www-form-urlencoded",
|
|
"Accept": "text/plain"}
|
|
conn = httplib.HTTPConnection("bugs.python.org")
|
|
print params, headers, conn
|
|
|
|
conn.request("POST", "", params, headers)
|
|
response = conn.getresponse()
|
|
print response.status, response.reason
|
|
|
|
data = response.read()
|
|
print data
|
|
|
|
conn.close()
|
|
|
|
"""
|
|
|
|
import cherrypy
|
|
import httplib, urllib
|
|
import sys
|
|
import unittest
|
|
|
|
import protocols.https.controller as controller
|
|
import santiago
|
|
|
|
|
|
class RestMonitor(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
super(RestMonitor, self).__init__()
|
|
|
|
self.monitor = controller.RestMonitor(None)
|
|
|
|
def test_valid_template_path(self):
|
|
"""Make sure the template path ends in a slash.
|
|
|
|
Otherwise, we can't find templates and the engine gets mighty buggered.
|
|
|
|
"""
|
|
self.assertTrue(self.monitor.relative_path.endswith("/"))
|
|
|
|
if sys.version_info < (2, 7):
|
|
"""Add a poor man's forward compatibility."""
|
|
|
|
class ContainsError(AssertionError):
|
|
pass
|
|
|
|
def assertIn(self, a, b):
|
|
if not a in b:
|
|
raise self.ContainsError("%s not in %s" % (a, b))
|
|
|
|
class MonitorTest(RestMonitor):
|
|
"""Generic test-class."""
|
|
|
|
def setUp(self):
|
|
self.santiago = santiago.Santiago(None)
|
|
|
|
self.client = 0
|
|
self.host = 1
|
|
self.service = 2
|
|
self.location = 3
|
|
|
|
self.monitor = controller.Monitor(santiago)
|
|
|
|
self.dispatcher = cherrypy.dispatch.RoutesDispatcher()
|
|
|
|
self.root = controller.Root(self.santiago)
|
|
cherrypy.tree.mount(self.root, "",
|
|
{"/": {"request.dispatch": self.dispatcher}})
|
|
|
|
def connect(self, routing_pairs):
|
|
for location, handler in routing_pairs:
|
|
self.monitor.rest_connect(self.dispatcher, location, handler)
|
|
|
|
def test_put(self):
|
|
raise NotImplementedError("MonitorTest.test_put")
|
|
|
|
def test_delete(self):
|
|
raise NotImplementedError("MonitorTest.test_delete")
|
|
|
|
class HostingTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.connect(('/hosting', controller.Hosting(self.santiago)))
|
|
|
|
def test_put(self):
|
|
params = urllib.urlencode({'put': self.client})
|
|
headers = {}
|
|
conn = httplib.HTTPConnection("https://localhost:8080/hosting/")
|
|
print params, headers, conn
|
|
|
|
conn.request("POST", "", params, headers)
|
|
response = conn.getresponse()
|
|
print response.status, response.reason
|
|
|
|
data = response.read()
|
|
print data
|
|
|
|
conn.close()
|
|
|
|
self.assertIn(self.santiago.hosting, self.client)
|
|
|
|
class HostingServicesTest(unittest.TestCase):
|
|
pass
|
|
class HostingLocationsTest(unittest.TestCase):
|
|
pass
|
|
class ClientTest(unittest.TestCase):
|
|
pass
|
|
class ClientServicesTest(unittest.TestCase):
|
|
pass
|
|
class ClientLocationsTest(unittest.TestCase):
|
|
pass
|
|
|
|
# class SantiagoTest(unittest.TestCase):
|
|
# """The base class for tests."""
|
|
#
|
|
# def setUp(self):
|
|
# super(TestServing, self).setUp()
|
|
#
|
|
# port_a = "localhost:9000"
|
|
# port_b = "localhost:8000"
|
|
#
|
|
# listeners_a = [santiago.SantiagoListener(port_a)]
|
|
# senders_a = [santiago.SantiagoSender()]
|
|
# listeners_b = [santiago.SantiagoListener(port_b)]
|
|
# senders_b = [santiago.SantiagoSender()]
|
|
#
|
|
# hosting_a = { "b": { "santiago": [ port_a ]}}
|
|
# consuming_a = { "santiagao": { "b": [ port_b ]}}
|
|
#
|
|
# hosting_b = { "a": { "santiago": [ port_b ],
|
|
# "wiki": [ "localhost:8001" ]}}
|
|
# consuming_b = { "santiagao": { "a": [ port_a ]}}
|
|
#
|
|
# self.santiago_a = Santiago(listeners_a, senders_a, hosting_a, consuming_a)
|
|
# self.santiago_b = Santiago(listeners_b, senders_b, hosting_b, consuming_b)
|
|
#
|
|
# def serveOnPort(self, port):
|
|
# """Start listening for connections on a named port.
|
|
#
|
|
# Used in testing as a mock listener for responses from a Santiago server.
|
|
#
|
|
# """
|
|
# class RequestReceiver(object):
|
|
# """A very basic listener.
|
|
#
|
|
# It merely records the calling arguments.
|
|
#
|
|
# """
|
|
# @cherrypy.expose
|
|
# def index(self, *args, **kwargs):
|
|
# self.args = args
|
|
# self.kwargs = kwargs
|
|
#
|
|
# self.socket_port = port
|
|
#
|
|
# self.receiver = RequestReceiver()
|
|
#
|
|
# cherrypy.quickstart(self.receiver)
|
|
#
|
|
# if sys.version_info < (2, 7):
|
|
# """Add a poor man's forward compatibility."""
|
|
#
|
|
# class ContainsError(AssertionError):
|
|
# pass
|
|
#
|
|
# def assertIn(self, a, b):
|
|
# if not a in b:
|
|
# raise self.ContainsError("%s not in %s" % (a, b))
|
|
#
|
|
# class TestClientInitialRequest(SantiagoTest):
|
|
# """Does the client send a correctly formed request?
|
|
#
|
|
# In these tests, we're sending requests to a mock listener which merely
|
|
# records that the requests were well-formed.
|
|
#
|
|
# """
|
|
# def setUp(self):
|
|
# super(SantiagoTest, self).setUp()
|
|
#
|
|
# self.serveOnPort(8000)
|
|
#
|
|
# def test_request(self):
|
|
# """Verify that A queues a properly formatted initial request."""
|
|
#
|
|
# self.santiago_a.request(from_="a", to="b",
|
|
# client="a", host="b",
|
|
# service="wiki", reply_to="localhost:9001")
|
|
#
|
|
# self.assertEqual(self.santiago_a.outgoing_messages,
|
|
# [{ "from": "a", "to": "b",
|
|
# "client": "a", "host": "b",
|
|
# "service": "wiki", "reply-to": "localhost:9001"}])
|
|
#
|
|
# def test_request(self):
|
|
# """Verify that A sends out a properly formatted initial request."""
|
|
#
|
|
# self.santiago_a.request(from_="a", to="b",
|
|
# client="a", host="b",
|
|
# service="wiki", reply_to="localhost:9001")
|
|
#
|
|
# self.santiago_a.process()
|
|
#
|
|
# self.assertEqual(self.receiver.kwargs,
|
|
# [{ "from": "a", "to": "b",
|
|
# "client": "a", "host": "b",
|
|
# "service": "wiki", "reply-to": "localhost:9001"}])
|
|
#
|
|
# class TestServerInitialRequest(SantiagoTest):
|
|
# """Test how the Santiago server replies to initial service requests.
|
|
#
|
|
# TODO: Add a mock listener to represent A.
|
|
# TODO: Transform the data structure tests into the mock-response tests.
|
|
# TODO tests: (normal serving + proxying) * (learning santiagi + not learning)
|
|
#
|
|
# Proxying
|
|
# ~~~~~~~~
|
|
#
|
|
# A host/listener (B) trusts proxied requests according to the minimum trust
|
|
# in the request. If the request comes from an untrusted proxy or is for an
|
|
# untrusted client, B ignores it.
|
|
#
|
|
# """
|
|
# def setUp(self):
|
|
# super(SantiagoTest, self).setUp()
|
|
#
|
|
# self.serveOnPort(9000)
|
|
#
|
|
# def test_acknowledgement(self):
|
|
# """If B receives an authorized request, then it replies with a location.
|
|
#
|
|
# An "authorized request" in this case is for a service from a client that
|
|
# B is willing to host that service for.
|
|
#
|
|
# In this case, B will answer with the wiki's location.
|
|
#
|
|
# """
|
|
# self.santiago_b.receive(from_="a", to="b",
|
|
# client="a", host="b",
|
|
# service="wiki", reply_to=None)
|
|
#
|
|
# self.assertEqual(self.santiago_b.outgoing_messages,
|
|
# [{"from": "b",
|
|
# "to": "a",
|
|
# "client": "a",
|
|
# "host": "b",
|
|
# "service": "wiki",
|
|
# "locations": ["192.168.0.13"],
|
|
# "reply-to": "localhost:8000"}])
|
|
#
|
|
# def test_reject_bad_service(self):
|
|
# """Does B reject requests for unsupported services?
|
|
#
|
|
# In this case, B should reply with an empty list of locations.
|
|
#
|
|
# """
|
|
# self.santiago_b.receive(from_="a", to="b",
|
|
# client="a", host="b",
|
|
# service="wiki", reply_to=None)
|
|
#
|
|
# self.assertEqual(self.santiago_b.outgoing_messages,
|
|
# [{"from": "b",
|
|
# "to": "a",
|
|
# "client": "a",
|
|
# "host": "b",
|
|
# "service": "wiki",
|
|
# "locations": [],
|
|
# "reply-to": "localhost:8000"}])
|
|
#
|
|
# def test_reject_bad_key(self):
|
|
# """If B receives a request from an unauthorized key, it does not reply.
|
|
#
|
|
# An "unauthorized request" in this case is for a service from a client
|
|
# that B does not trust. This is different than clients B hosts no
|
|
# services for.
|
|
#
|
|
# In this case, B will never answer the request.
|
|
#
|
|
# """
|
|
# self.santiago_b.receive(from_="a", to="b",
|
|
# client="z", host="b",
|
|
# service="wiki", reply_to=None)
|
|
#
|
|
# self.assertEqual(self.santiago_b.outgoing_messages, [])
|
|
#
|
|
# def test_reject_good_source_bad_client(self):
|
|
# """B is silent when a trusted key proxies anything for an untrusted key.
|
|
#
|
|
# B doesn't know who the client is and should consider it an
|
|
# untrusted key connection attempt.
|
|
#
|
|
# """
|
|
# self.santiago_b.receive(from_="a", to="b",
|
|
# client="z", host="b",
|
|
# service="wiki", reply_to=None)
|
|
#
|
|
# self.assertEqual(self.santiago_b.outgoing_messages, [])
|
|
#
|
|
# def test_reject_bad_source_good_client(self):
|
|
# """B is silent when an untrusted key proxies anything for a trusted key.
|
|
#
|
|
# B doesn't know who the proxy is and should consider it an
|
|
# untrusted key connection attempt.
|
|
#
|
|
# """
|
|
# self.santiago_b.receive(from_="z", to="b",
|
|
# client="a", host="b",
|
|
# service="wiki", reply_to=None)
|
|
#
|
|
# self.assertEqual(self.santiago_b.outgoing_messages, [])
|
|
#
|
|
# def test_reject_bad_source_bad_client(self):
|
|
# """B is silent when untrusted keys proxy anything for untrusted keys.
|
|
#
|
|
# B doesn't know who anybody is and considers this an untrusted
|
|
# connection attempt.
|
|
#
|
|
# """
|
|
# self.santiago_b.receive(from_="y", to="b",
|
|
# client="z", host="b",
|
|
# service="wiki", reply_to=None)
|
|
#
|
|
# self.assertEqual(self.santiago_b.outgoing_messages, [])
|
|
#
|
|
# def test_learn_santaigo(self):
|
|
# """Does B learn new Santiago locations from trusted requests?
|
|
#
|
|
# If A sends B a request with a new Santiago location, B should learn it.
|
|
#
|
|
# """
|
|
# self.santiago_b.receive(from_="a", to="b",
|
|
# client="a", host="b",
|
|
# service="wiki", reply_to="localhost:9001")
|
|
#
|
|
# self.assertEqual(self.santiago_b.consuming["santiago"]["a"],
|
|
# ["localhost:9000", "localhost:9001"])
|
|
#
|
|
# def test_handle_requests_once(self):
|
|
# """Verify that we reply to each request only once."""
|
|
#
|
|
# self.santiago_b.receive(from_="a", to="b",
|
|
# client="a", host="b",
|
|
# service="wiki", reply_to=None)
|
|
# self.santiago_b.process()
|
|
#
|
|
# self.assertEqual(self.santiago_b.outgoing_messages, [])
|
|
#
|
|
# class TestServerInitialResponse(SantiagoTest):
|
|
# pass
|
|
#
|
|
# class TestClientInitialResponse(SantiagoTest):
|
|
# pass
|
|
#
|
|
# class TestForwardedRequest(SantiagoTest):
|
|
# pass
|
|
#
|
|
# class TestForwardedResponse(SantiagoTest):
|
|
# pass
|
|
#
|
|
# class TestSimpleSantiago(unittest.TestCase):
|
|
# def setUp(self):
|
|
#
|
|
# port_a = "localhost:9000"
|
|
# port_b = "localhost:8000"
|
|
#
|
|
# listeners_a = {"http": {"port": port_a}}
|
|
# senders_a = ({ "protocol": "http", "proxy": tor_proxy_port },)
|
|
#
|
|
# listeners_b = {"http": {"port": port_b}}
|
|
# senders_b = ({ "protocol": "http", "proxy": tor_proxy_port },)
|
|
#
|
|
# hosting_a = { "b": { "santiago": set( ["aDifferentHexNumber.onion"])}}
|
|
# consuming_a = { "santiagao": {"b": set(["iAmAHexadecimalNumber.onion"])}}
|
|
#
|
|
# hosting_b = { "a": { "santiago": set( ["iAmAHexadecimalNumber.onion"])}}
|
|
# consuming_b = { "santiagao": { "a": set( ["aDifferentHexNumber.onion"])}}
|
|
#
|
|
# self.santiago_a = santiago.Santiago(listeners_a, senders_a,
|
|
# hosting_a, consuming_a, "a")
|
|
# self.santiago_b = santiago.Santiago(listeners_b, senders_b,
|
|
# hosting_b, consuming_b, "b")
|
|
#
|
|
# cherrypy.Application(self.santiago_a, "/")
|
|
# cherrypy.Application(self.santiago_b, "/")
|
|
#
|
|
# cherrypy.engine.start()
|
|
#
|
|
# def testRequest(self):
|
|
# self.santiago_a.request(from_="a", to="b",
|
|
# client="a", host="b",
|
|
# service="wiki", reply_to="localhost:9000")
|
|
#
|
|
#
|
|
# class Unwrapping(unittest.TestCase):
|
|
#
|
|
# def testVerifySigner(self):
|
|
# pass
|
|
#
|
|
# def testVerifyClient(self):
|
|
# pass
|
|
#
|
|
# def testDecryptClient(self):
|
|
# pass
|
|
#
|
|
# class IncomingProxyRequest(unittest.TestCase):
|
|
#
|
|
# """Do we correctly handle valid, incoming, proxied messages?
|
|
#
|
|
# These tests are for the first wrapped layer of the message, that which is
|
|
# signed by the sender. The sender is not necessarily the original requester
|
|
# who's asking us to do something with the message.
|
|
#
|
|
# """
|
|
#
|
|
# def setUp(self):
|
|
# pass
|
|
#
|
|
# def testPassingMessage(self):
|
|
# """Does a valid proxied message pass?"""
|
|
#
|
|
# pass
|
|
#
|
|
# def testInvalidSig(self):
|
|
# """Does an invalid signature raise an error?"""
|
|
#
|
|
# pass
|
|
#
|
|
# def testUnknownClient(self):
|
|
# """Does an unknown client raise an error?"""
|
|
#
|
|
# pass
|
|
#
|
|
# class IncomingSignedRequest(IncomingProxyRequest):
|
|
#
|
|
# """Do we correctly handle valid, incoming, messages?
|
|
#
|
|
# These tests focus on the second layer of the message which is signed by the
|
|
# host/client and lists a destination.
|
|
#
|
|
# """
|
|
# def testProxyOtherHosts(self):
|
|
# """Messages to others are sent to them directly or proxied."""
|
|
#
|
|
# pass
|
|
#
|
|
# def testHandleMyHosting(self):
|
|
# """Messages to me are not proxied and handled normally."""
|
|
#
|
|
# pass
|
|
#
|
|
# def testNoDestination(self):
|
|
# """Messages without destinations are ignored."""
|
|
#
|
|
# pass
|
|
#
|
|
# class IncomingRequestBody(IncomingSignedRequest):
|
|
#
|
|
# """Do we correctly handle the body of a request?
|
|
#
|
|
# This is the last layer of the message which is encrypted by the original
|
|
# sender. This validation also depends on the previous layer's data, making
|
|
# it a bit more complicated.
|
|
#
|
|
# """
|
|
# def testHandleGoodMessage(self):
|
|
# """Sanity check: no errors are thrown for a valid message."""
|
|
#
|
|
# pass
|
|
#
|
|
# def testCantDecryptMessage(self):
|
|
# """This message isn't for me. I can't decrypt it."""
|
|
#
|
|
# pass
|
|
#
|
|
# def testImNotHost(self):
|
|
# """Bail out if someone else is the host, yet I am the "to"."""
|
|
#
|
|
# pass
|
|
#
|
|
# def testImNotClient(self):
|
|
# """Bail out if someone else is the client, yet I am the "to"."""
|
|
#
|
|
# pass
|
|
#
|
|
# def testHostAndClient(self):
|
|
# """Bail out if the message includes a host and a client.
|
|
#
|
|
# A circular response?
|
|
#
|
|
# """
|
|
# pass
|
|
#
|
|
# def testImNotTo(self):
|
|
# """This message isn't for me.
|
|
#
|
|
# The "To" has been repeated from the signed message, but I'm not the
|
|
# recipient in the encrypted message.
|
|
#
|
|
# """
|
|
# pass
|
|
#
|
|
# def testNoDestinations(self):
|
|
# """No host, client, or to."""
|
|
#
|
|
# pass
|
|
#
|
|
# def testSignersDiffer(self):
|
|
# """The signed message and encrypted message have different signers."""
|
|
#
|
|
# pass
|
|
#
|
|
# def testSignerAndClientDiffer(self):
|
|
# """The encrypted message is signed by someone other than the cilent."""
|
|
#
|
|
# pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|