medashare/ui/server.py

#!/usr/bin/env python

# Notes:
# Python requests: https://2.python-requests.org/en/master/
# IRequest interface: https://twistedmatrix.com/documents/current/api/twisted.web.iweb.IRequest.html
# IResource interface: https://twistedmatrix.com/documents/current/api/twisted.web.resource.IResource.html
# Twisted TDD: https://twistedmatrix.com/documents/current/core/howto/trial.html
# Hypothesis: https://hypothesis.readthedocs.io/en/latest/
# Going Async from Flask to Twisted Klein: https://crossbario.com/blog/Going-Asynchronous-from-Flask-to-Twisted-Klein/
# Klein POST docs: https://klein.readthedocs.io/en/latest/examples/handlingpost.html

from contextlib import nested
from klein import Klein
from kleintest import *
from twisted.trial import unittest
from twisted.web.iweb import IRequest
from cli import Persona, MDBase

import hashlib
import mock
import os.path
import shutil
import tempfile

defaultfile = 'mediaserver.store.pasn1'
class MEDAServer:
	def __init__(self, fname):
		self._trustedkeys = {}

	app = Klein()

	def addpubkey(self, pubkey):
		persona = Persona.from_pubkey(pubkey)

		self._trustedkeys[persona.uuid] = persona

	def store(self):
		pass

	@app.route('/')
	def home(request):
		return 'hello'

	@app.route('/store')
	def storeobj(self, request):
		try:
			obj = MDBase.decode(request.content.read())

			#if obj.type == 'identity':
			keyuuid = obj.uuid
			#else:
			#	keyuuid = obj.created_by_ref

			persona = self._trustedkeys[keyuuid]
			persona.verify(obj)

			request.setResponseCode(201)

		except Exception:
			request.setResponseCode(401)

# twistd support
#medaserver = MEDAServer()
#resource = medaserver.app.resource

def main():
	from optparse import OptionParser

	parser = OptionParser()
	parser.add_option('-a', action='append', dest='addpubkey',
	    default=[], help='Add specified public key as a trusted key.')

	options, args = parser.parse_args()

	medaserver = MEDAServer(defaultfile)

	try:
		if options.addpubkey:
			for i in options.addpubkey:
				medaserver.addpubkey(i)
			return

		medaserver.app.run()
	finally:
		medaserver.store()

if __name__ == '__main__':	# pragma: no cover
	main()

class _TestCases(unittest.TestCase):
	def setUp(self):
		d = os.path.realpath(tempfile.mkdtemp())
		self.basetempdir = d
		self.medaserverfile = os.path.join(self.basetempdir, 'serverstore.pasn1')
		self.medaserver = MEDAServer(self.medaserverfile)
		self.requests = FakeRequests(self.medaserver.app)

	def tearDown(self):
		shutil.rmtree(self.basetempdir)
		self.basetempdir = None

	def test_404(self):
		h = hashlib.sha256(open('fixtures/testfiles/test.txt').read()).hexdigest()
		r = self.requests.get('/chash/%s' % h)
		self.assertEqual(r.status_code, 404)

	def test_pubkeystorage(self):
		import cli

		# that an identity
		persona = cli.Persona()
		persona.generate_key()

		# that by default, put's
		r = self.requests.put('/store', data=persona.get_identity().encode())

		# are denied
		self.assertEqual(r.status_code, 401)

		# can have it's public key added to the server
		self.medaserver.addpubkey(persona.get_pubkey())

		# that it can store the pubkey's identity
		r = self.requests.put('/store', data=persona.get_identity().encode())
		self.assertEqual(r.status_code, 201)

		# that an object with a bad signature
		badsigobj = persona.get_identity().new_version()

		# when stored
		r = self.requests.put('/store', data=badsigobj.encode())

		# is rejected
		self.assertEqual(r.status_code, 401)

		# that when stored
		self.medaserver.store()

		tmpmedaserver = MEDAServer(self.medaserverfile)

	@mock.patch('klein.Klein.run')
	def test_addpubkey(self, apprun):
		import cli

		persona = cli.Persona()
		persona.generate_key()

		with nested(mock.patch('server.MEDAServer.addpubkey'),
		    mock.patch('sys.argv', [ 'progname', '-a',
		    persona.get_pubkey() ])) as (addpub, argv):
			main()

		addpub.assert_called_with(persona.get_pubkey())
		apprun.assert_not_called()

	# Note: because of this mock, it hides the actual app.run call w/
	# a mock
	@mock.patch('server.MEDAServer')
	def test_medaserverinstanciated(self, medaserver):
		# that when main is run
		main()

		# that it gets called with the default storage file
		medaserver.assert_called_with('mediaserver.store.pasn1')

	@mock.patch('server.MEDAServer.store')
	@mock.patch('klein.Klein.run')
	def test_appruns(self, kleinrun, storefun):
		main()

		kleinrun.assert_called()
		storefun.assert_called()