mikitsu · February 16, 2021 13:21
diff --git a/simple_gemini_client.py b/simple_gemini_client.py
 #!/usr/bin/env python3
 import ssl
 import sys
 import cgi
 import socket
 import hashlib
 import urllib.parse
 import json

 MAX_RECV_CHUNKS = 10 * 1024  #  == 10 MiB


 class Client:
    def __init__(self, tofu_certs):
        self.tofu_certs = tofu_certs
        if sys.version_info >= (3, 7):
            self.ssl_context = ssl.SSLContext()
            self.ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2
        else:
            self.ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
        self._base = 'gemini://'
        self.links = []

    def visit(self, url: str):
        try:
            self.handle_response(self.make_request(
                self.links[int(url) - 1] if url.isdigit() else url))
        except OSError as e:
            print('internal ERROR:', type(e).__name__, e)

    def make_request(self, user_url: str) -> bytes:
        # urllib.parse.urljoin doesn't know about gemini (7 == len('gemini:'))
        if user_url.startswith('?') and '?' in self._base:
            url = self._base + '&' + user_url[1:]
        else:
            url = urllib.parse.urljoin(
                self._base[self._base.startswith('gemini:')*7:], user_url)
        url = 'gemini:'*url.startswith('//') + url
        netloc = urllib.parse.urlparse(url).netloc
        host, port = netloc.split(':') if ':' in netloc else [netloc, 1965]
        sock = socket.socket()
        sock = self.ssl_context.wrap_socket(sock, server_hostname=host)
        sock.connect((host, int(port)))
        cert_fp = hashlib.sha256(sock.getpeercert(binary_form=True)).hexdigest()
        if self.tofu_certs.setdefault(host, cert_fp) != cert_fp:
            raise ssl.SSLError('TOFU hash match failed')  # best error type?
        sock.sendall(url.encode() + b'\r\n')
        data = b''.join(sock.recv(1024) for __ in range(MAX_RECV_CHUNKS))
        sock.close()
        self._base = url
        return data

    def handle_response(self, response_data: bytes):
        status, body = response_data.split(b'\r\n', 1)
        code, meta = status.decode().split(' ', 1)
        if code[0] == '1':
            self.visit('?' + urllib.parse.quote(input(meta + ' -> ')))
        elif code[0] in '45':
            print('ERROR code from server:', meta)
        elif code[0] == '3':
            self.visit(meta)
        elif code[0] == '2':
            mime_type, params = cgi.parse_header(meta)
            if not mime_type or mime_type == 'text/gemini':
                self.display_gmi(body.decode(params.get('charset', 'utf-8')))
            elif mime_type.startswith('text/'):
                print(body.decode(params.get('charset', 'utf-8')))
            else:
                print('unsupported MIME type:', mime_type)
        else:
            print("can't handle server response code:", code)

    def display_gmi(self, text):
        self.links = []
        pre = False
        for line in text.splitlines():
            if line.startswith('```'):
                pre = not pre
            elif not pre and line.startswith('=> '):
                __, link, *display = line.split(maxsplit=2)
                self.links.append(link)
                print(len(self.links), '=>', *display, f'[{link}]')
            else:  # both pre-formatted and normal, hope the terminal wraps
                print(line)


 if __name__ == '__main__':
    sys.argv.append('/dev/null')  # default if no TOFU file specified
    try:
        with open(sys.argv[1]) as f:
            client = Client(json.load(f))
    except (json.JSONDecodeError, FileNotFoundError):
        client = Client({})
    try:
        while True:
            client.visit(input('url or link -> '))
    except (KeyboardInterrupt, EOFError):
        print('Bye.')
    with open(sys.argv[1], 'w') as f:
        json.dump(client.tofu_certs, f)
diff --git a/usage.md b/usage.md
	#!/usr/bin/env python3
	import ssl
	import sys
	import cgi
	import socket
	import hashlib
	import urllib.parse
	import json

	MAX_RECV_CHUNKS = 10 * 1024 # == 10 MiB


	class Client:
	def __init__(self, tofu_certs):
	self.tofu_certs = tofu_certs
	if sys.version_info >= (3, 7):
	self.ssl_context = ssl.SSLContext()
	self.ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2
	else:
	self.ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
	self._base = 'gemini://'
	self.links = []

	def visit(self, url: str):
	try:
	self.handle_response(self.make_request(
	self.links[int(url) - 1] if url.isdigit() else url))
	except OSError as e:
	print('internal ERROR:', type(e).__name__, e)

	def make_request(self, user_url: str) -> bytes:
	# urllib.parse.urljoin doesn't know about gemini (7 == len('gemini:'))
	if user_url.startswith('?') and '?' in self._base:
	url = self._base + '&' + user_url[1:]
	else:
	url = urllib.parse.urljoin(
	self._base[self._base.startswith('gemini:')*7:], user_url)
	url = 'gemini:'*url.startswith('//') + url
	netloc = urllib.parse.urlparse(url).netloc
	host, port = netloc.split(':') if ':' in netloc else [netloc, 1965]
	sock = socket.socket()
	sock = self.ssl_context.wrap_socket(sock, server_hostname=host)
	sock.connect((host, int(port)))
	cert_fp = hashlib.sha256(sock.getpeercert(binary_form=True)).hexdigest()
	if self.tofu_certs.setdefault(host, cert_fp) != cert_fp:
	raise ssl.SSLError('TOFU hash match failed') # best error type?
	sock.sendall(url.encode() + b'\r\n')
	data = b''.join(sock.recv(1024) for __ in range(MAX_RECV_CHUNKS))
	sock.close()
	self._base = url
	return data

	def handle_response(self, response_data: bytes):
	status, body = response_data.split(b'\r\n', 1)
	code, meta = status.decode().split(' ', 1)
	if code[0] == '1':
	self.visit('?' + urllib.parse.quote(input(meta + ' -> ')))
	elif code[0] in '45':
	print('ERROR code from server:', meta)
	elif code[0] == '3':
	self.visit(meta)
	elif code[0] == '2':
	mime_type, params = cgi.parse_header(meta)
	if not mime_type or mime_type == 'text/gemini':
	self.display_gmi(body.decode(params.get('charset', 'utf-8')))
	elif mime_type.startswith('text/'):
	print(body.decode(params.get('charset', 'utf-8')))
	else:
	print('unsupported MIME type:', mime_type)
	else:
	print("can't handle server response code:", code)

	def display_gmi(self, text):
	self.links = []
	pre = False
	for line in text.splitlines():
	if line.startswith('```'):
	pre = not pre
	elif not pre and line.startswith('=> '):
	__, link, *display = line.split(maxsplit=2)
	self.links.append(link)
	print(len(self.links), '=>', *display, f'[{link}]')
	else: # both pre-formatted and normal, hope the terminal wraps
	print(line)


	if __name__ == '__main__':
	sys.argv.append('/dev/null') # default if no TOFU file specified
	try:
	with open(sys.argv[1]) as f:
	client = Client(json.load(f))
	except (json.JSONDecodeError, FileNotFoundError):
	client = Client({})
	try:
	while True:
	client.visit(input('url or link -> '))
	except (KeyboardInterrupt, EOFError):
	print('Bye.')
	with open(sys.argv[1], 'w') as f:
	json.dump(client.tofu_certs, f)