💾 Archived View for gemini.dimakrasner.com › main.py captured on 2023-04-19 at 22:37:45. Gemini links have been rewritten to link to archived content
⬅️ Previous capture (2023-03-20)
-=-=-=-=-=-=-
import ussl
import socket
import network
import _thread
import machine
import urequests
import time
import zlib
import _thread
import uselect
import time
from conf import *
keyfile = open("key.der", "rb").read()
certfile = open("cert.der", "rb").read()
py_status = b'20 text/x-script.python\r\n'
gmi_status = b'20 text/gemini\r\n'
def get_path(req):
url = req[:-2]
print(url)
try:
_, _, _, path = url.split(b'/', 3)
except ValueError:
_, _, _, path = (url + b'/').split(b'/', 3)
path = path.strip(b'/')
if len(path) == 0:
return 'index.gmi'
return path.decode('utf-8')
def update_public_ip():
print('Updating public address')
try:
urequests.get(f"https://www.duckdns.org/update?domains={DOMAINS}&token={TOKEN}")
except Exception as e:
print(e)
def update_public_ip_thread():
while True:
time.sleep(86400)
update_public_ip()
class Request:
def __init__(self, ident, s, poller):
self.id = ident
self.start = time.ticks_ms()
self.deadline = time.ticks_add(self.start, 20000)
self.s = s
self.poller = poller
self.ssl = None
self.req = bytearray(64)
self.req_off = 0
self.status = None
self.status_off = 0
self.chunk = None
self.chunk_off = 0
self.f = None
poller.register(s, uselect.POLLIN)
def closenotify(self):
self.ssl.closenotify()
def close(self):
if self.ssl:
self.poller.unregister(self.ssl)
self.ssl.close()
if self.s:
self.poller.unregister(self.s)
self.s.close()
if self.f:
self.f.close()
time.sleep(10)
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(SSID, PASSPHRASE)
while wlan.ifconfig()[0] == '0.0.0.0':
print('Waiting for connection')
time.sleep(1)
s = socket.socket()
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.setblocking(False)
s.bind(('0.0.0.0', PORT))
s.listen(5)
led = machine.Pin('LED', machine.Pin.OUT)
print(f"Local address: {wlan.ifconfig()[0]}")
update_public_ip()
poller = uselect.poll()
poller.register(s, uselect.POLLIN)
next_id = 0
requests = []
print('Ready')
while True:
poller.register(s, uselect.POLLIN)
now = time.ticks_ms()
res = poller.poll(1000)
led.value(1 if res or requests else 0)
new_conn = False
events = {}
for t in res:
strm = t[0]
event = t[1]
if strm is s:
new_conn = True
else:
reqs = [req for req in requests if req.s is strm or req.ssl is strm]
if len(reqs) > 1:
raise Exception("Invalid state")
req = reqs[0]
if not req:
print(f"Unknown stream: {strm}")
continue
events[req] = event
for i in range(2):
for req in requests:
try:
if i == 0:
if now >= req.deadline:
raise Exception(f"{req.id} has timed out")
continue
event = events.get(req)
if event is None:
continue
if event == uselect.POLLHUP or event == uselect.POLLHUP:
raise Exception(f"EOF from {req.id}")
if event == uselect.POLLOUT:
if req.status_off < len(req.status):
print(f"Sending status line to {req.id}")
sent = req.ssl.write(req.status)
if sent <= 0:
raise Exception(f"Failed to send status line to {req.id}")
req.status_off += sent
if not req.f:
try:
req.closenotify()
finally:
raise Exception(f"Done sending status line to {req.id}")
else:
if not req.chunk or req.chunk_off >= len(req.chunk):
req.chunk = req.f.read(1024)
if not req.chunk:
try:
req.closenotify()
finally:
raise Exception(f"Done sending response to {req.id}")
req.chunk_off = 0
print(f"Sending {len(req.chunk) - req.chunk_off} to {req.id}")
sent = req.ssl.write(req.chunk[req.chunk_off:])
if sent <= 0:
raise Exception(f"Failed to send response to {req.id}")
req.chunk_off += sent
continue
if req.ssl is None:
print(f"Handshake on {req.id}")
req.ssl = ussl.wrap_socket(req.s, server_side=True, key=keyfile, cert=certfile, do_handshake=False)
poller.unregister(req.s)
poller.register(req.ssl, uselect.POLLIN)
continue
b = req.ssl.read(1)
if b is None:
continue
if len(b) == 0:
raise Exception(f"EOF from {req.id}")
req.req[req.req_off] = b[0]
req.req_off += 1
if req.req_off == 9 and req.req[:req.req_off] != b'gemini://':
raise Exception(f"Invalid scheme from {req.id}: {req.req[:req.req_off]}")
# b'gemini://a.b\r\n'
if req.req_off < 14:
continue
eof = req.req[:req.req_off].endswith(b'\r\n')
if not eof and req.req_off >= len(req.req):
raise Exception(f"Invalid request from {req.id}: {req.req[:req.req_off]}")
if not eof:
continue
path = get_path(req.req[:req.req_off])
if '/' in path or path == 'key.der' or path == 'cert.der' or path == 'conf.py' or path == 'boot.py':
print(f"Forbidden path from {req.id}: {path}")
req.resp = b'40 Forbidden\r\n'
else:
print(f"Serving {path} to {req.id}")
try:
req.f = open(path, "rb")
req.status = py_status if path == b'main.py' else gmi_status
except Exception as e:
print(e)
req.status = b'41 Internal server error\r\n'
poller.modify(req.ssl, uselect.POLLOUT)
except Exception as e:
print(e)
print(f"Closing {req.id}")
req.close()
if len(requests) == 5:
print('Allowing new connections')
poller.register(s, uselect.POLLIN)
requests.remove(req)
if new_conn:
try:
print('Accepting new connection')
c, _ = s.accept()
except Exception as e:
print(e)
continue
try:
c.setblocking(False)
except Exception as e:
print(e)
c.close()
continue
next_id += 1
req = Request(next_id, c, poller)
requests.append(req)
print(f"Accepted {req.id}, have {len(requests)} connections")
if len(requests) == 5:
print('Reached the maximum number of concurrent requests')
poller.unregister(s)