TonBERRY/microdot
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SSL/TLS Support

Browse Source
This commit is contained in:
Miguel Grinberg
2022-09-05 00:33:24 +01:00
parent 2399c29c8a
commit b61f51f243
9 changed files with 243 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@@ -2,3 +2,4 @@
omit=
src/microdot_websocket_alt.py
src/microdot_asgi_websocket.py
src/microdot_ssl.py

45
docs/extensions.rst
View File

@@ -282,6 +282,51 @@ but the ``receive()`` and ``send()`` methods are asynchronous.
`echo_asgi.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/websocket/echo_asgi.py>`_
example shows how to use this module.
HTTPS Support
~~~~~~~~~~~~~
.. list-table::
:align: left
* - Compatibility
- | CPython & MicroPython
* - Required Microdot source files
- | `microdot.py <https://github.com/miguelgrinberg/microdot/tree/main/src/microdot.py>`_
| `microdot_ssl.py <https://github.com/miguelgrinberg/microdot/tree/main/src/microdot_ssl.py>`_
* - Examples
- | `hello_tls.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/tls/hello_tls.py>`_
| `hello_asyncio_tls.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/tls/hello_asyncio_tls.py>`_
The ``run()`` function accepts an optional ``ssl`` argument, through which an
initialized ``SSLContext`` object can be passed. MicroPython does not currently
have a ``SSLContext`` implementation, so the ``microdot_ssl`` module provides
a basic implementation that can be used to create a context.
Example::
from microdot import Microdot
from microdot_ssl import create_ssl_context
app = Microdot()
@app.route('/')
def index(req):
return 'Hello, World!'
sslctx = create_ssl_context('cert.der', 'key.der')
app.run(port=4443, debug=True, ssl=sslctx)
.. note::
The ``microdot_ssl`` module is only needed for MicroPython. When used under
CPython, this module creates a standard ``SSLContext`` instance.
.. note::
The ``uasyncio`` library for MicroPython does not currently support TLS, so
this feature is not available for asynchronous applications on that
platform. The ``asyncio`` library for CPython is fully supported.
Test Client
~~~~~~~~~~~

20
examples/tls/README.md Normal file
View File

@@ -0,0 +1,20 @@
This directory contains examples that demonstrate how to start TLS servers.
To run these examples, SSL certificate and private key files need to be
created. When running under CPython, the files should be in PEM format, named
`cert.pem` and `key.pem`. When running under MicroPython, they should be in DER
format, and named `cert.der` and `key.der`.
To quickly create a self-signed SSL certificate, use the following command:
```bash
openssl req -x509 -newkey rsa:4096 -nodes -out cert.pem -keyout key.pem -days 365
```
To convert the resulting PEM files to DER format for MicroPython, use these
commands:
```bash
openssl x509 -in localhost.pem -out localhost.der -outform DER
openssl rsa -in localhost-key.pem -out localhost-key.der -outform DER
```

35
examples/tls/hello_async_tls.py Normal file
View File

@@ -0,0 +1,35 @@
import ssl
from microdot_asyncio import Microdot
app = Microdot()
htmldoc = '''<!DOCTYPE html>
<html>
<head>
<title>Microdot Example Page</title>
</head>
<body>
<div>
<h1>Microdot Example Page</h1>
<p>Hello from Microdot!</p>
<p><a href="/shutdown">Click to shutdown the server</a></p>
</div>
</body>
</html>
'''
@app.route('/')
async def hello(request):
return htmldoc, 200, {'Content-Type': 'text/html'}
@app.route('/shutdown')
async def shutdown(request):
request.app.shutdown()
return 'The server is shutting down...'
sslctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
sslctx.load_cert_chain('cert.pem', 'key.pem')
app.run(port=4443, debug=True, ssl=sslctx)

36
examples/tls/hello_tls.py Normal file
View File

@@ -0,0 +1,36 @@
import sys
from microdot import Microdot
from microdot_ssl import create_ssl_context
app = Microdot()
htmldoc = '''<!DOCTYPE html>
<html>
<head>
<title>Microdot Example Page</title>
</head>
<body>
<div>
<h1>Microdot Example Page</h1>
<p>Hello from Microdot!</p>
<p><a href="/shutdown">Click to shutdown the server</a></p>
</div>
</body>
</html>
'''
@app.route('/')
def hello(request):
return htmldoc, 200, {'Content-Type': 'text/html'}
@app.route('/shutdown')
def shutdown(request):
request.app.shutdown()
return 'The server is shutting down...'
ext = 'der' if sys.implementation.name == 'micropython' else 'pem'
sslctx = create_ssl_context('cert.' + ext, 'key.' + ext)
app.run(port=4443, debug=True, ssl=sslctx)

7
src/microdot.py
View File

@@ -861,7 +861,7 @@ class Microdot():
"""
raise HTTPException(status_code, reason)
def run(self, host='0.0.0.0', port=5000, debug=False):
def run(self, host='0.0.0.0', port=5000, debug=False, ssl=None):
"""Start the web server. This function does not normally return, as
the server enters an endless listening loop. The :func:`shutdown`
function provides a method for terminating the server gracefully.
@@ -877,6 +877,8 @@ class Microdot():
port 5000.
:param debug: If ``True``, the server logs debugging information. The
default is ``False``.
:param ssl: An ``SSLContext`` instance or ``None`` if the server should
not use TLS. The default is ``None``.
Example::
@@ -904,6 +906,9 @@ class Microdot():
self.server.bind(addr)
self.server.listen(5)
if ssl:
self.server = ssl.wrap_socket(self.server, server_side=True)
while not self.shutdown_requested:
try:
sock, addr = self.server.accept()

19
src/microdot_asyncio.py
View File

@@ -207,7 +207,8 @@ class Response(BaseResponse):
class Microdot(BaseMicrodot):
async def start_server(self, host='0.0.0.0', port=5000, debug=False):
async def start_server(self, host='0.0.0.0', port=5000, debug=False,
ssl=None):
"""Start the Microdot web server as a coroutine. This coroutine does
not normally return, as the server enters an endless listening loop.
The :func:`shutdown` function provides a method for terminating the
@@ -224,6 +225,8 @@ class Microdot(BaseMicrodot):
port 5000.
:param debug: If ``True``, the server logs debugging information. The
default is ``False``.
:param ssl: An ``SSLContext`` instance or ``None`` if the server should
not use TLS. The default is ``None``.
This method is a coroutine.
@@ -266,7 +269,12 @@ class Microdot(BaseMicrodot):
print('Starting async server on {host}:{port}...'.format(
host=host, port=port))
self.server = await asyncio.start_server(serve, host, port)
try:
self.server = await asyncio.start_server(serve, host, port,
ssl=ssl)
except TypeError:
self.server = await asyncio.start_server(serve, host, port)
while True:
try:
await self.server.wait_closed()
@@ -276,7 +284,7 @@ class Microdot(BaseMicrodot):
# wait a bit and try again
await asyncio.sleep(0.1)
def run(self, host='0.0.0.0', port=5000, debug=False):
def run(self, host='0.0.0.0', port=5000, debug=False, ssl=None):
"""Start the web server. This function does not normally return, as
the server enters an endless listening loop. The :func:`shutdown`
function provides a method for terminating the server gracefully.
@@ -292,6 +300,8 @@ class Microdot(BaseMicrodot):
port 5000.
:param debug: If ``True``, the server logs debugging information. The
default is ``False``.
:param ssl: An ``SSLContext`` instance or ``None`` if the server should
not use TLS. The default is ``None``.
Example::
@@ -305,7 +315,8 @@ class Microdot(BaseMicrodot):
app.run(debug=True)
"""
asyncio.run(self.start_server(host=host, port=port, debug=debug))
asyncio.run(self.start_server(host=host, port=port, debug=debug,
ssl=ssl))
def shutdown(self):
self.server.close()

61
src/microdot_ssl.py Normal file
View File

@@ -0,0 +1,61 @@
import ssl
def create_ssl_context(cert, key, **kwargs):
"""Create an SSL context to wrap sockets with.
:param cert: The certificate to use. If it is given as a string, it is
assumed to be a filename. If it is given as a bytes object, it
is assumed to be the certificate data. In both cases the data
is expected to be in PEM format for CPython and in DER format
for MicroPython.
:param key: The private key to use. If it is given as a string, it is
assumed to be a filename. If it is given as a bytes object, it
is assumed to be the private key data. in both cases the data
is expected to be in PEM format for CPython and in DER format
for MicroPython.
:param kwargs: Additional arguments to pass to the ``ssl.wrap_socket``
function.
Note: This function creates a fairly limited SSL context object to enable
the use of certificates under MicroPython. It is not intended to be used in
any other context, and in particular, it is not needed when using CPython
or any other Python implementation that has native support for
``SSLContext`` objects. Once MicroPython implements ``SSLContext``
natively, this function will be deprecated.
"""
if hasattr(ssl, 'SSLContext'):
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER, **kwargs)
ctx.load_cert_chain(cert, key)
return ctx
if isinstance(cert, str):
with open(cert, 'rb') as f:
cert = f.read()
if isinstance(key, str):
with open(key, 'rb') as f:
key = f.read()
class FakeSSLSocket:
def __init__(self, sock, **kwargs):
self.sock = sock
self.kwargs = kwargs
def accept(self):
client, addr = self.sock.accept()
return (ssl.wrap_socket(client, cert=cert, key=key, **self.kwargs),
addr)
def close(self):
self.sock.close()
class FakeSSLContext:
def __init__(self, **kwargs):
self.kwargs = kwargs
def wrap_socket(self, sock, **kwargs):
all_kwargs = self.kwargs.copy()
all_kwargs.update(kwargs)
return FakeSSLSocket(sock, **all_kwargs)
return FakeSSLContext(**kwargs)

24
tests/test_microdot.py
View File

@@ -570,3 +570,27 @@ class TestMicrodot(unittest.TestCase):
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'], 'text/plain')
self.assertEqual(res.text, 'before:foo:after')
def test_ssl(self):
self._mock()
app = Microdot()
@app.route('/foo')
def foo(req):
return 'bar'
class FakeSSL:
def wrap_socket(self, sock, **kwargs):
return sock
mock_socket.clear_requests()
fd = mock_socket.add_request('GET', '/foo')
self._add_shutdown(app)
app.run(ssl=FakeSSL())
self.assertTrue(fd.response.startswith(b'HTTP/1.0 200 OK\r\n'))
self.assertIn(b'Content-Length: 3\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\nbar'))
self._unmock()