yah2g: $URL

blob - /dev/null blob + e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 (mode 644) blob - /dev/null blob + 3bf2a7e59a678f38ffb841f709fa0c7f04a84072 (mode 644) --- /dev/null +++ yah2g.py @@ -0,0 +1,251 @@ +"""Yet another http-to-gemini.""" +import socket +import ssl +import xml.etree.ElementTree as ET +from argparse import ArgumentParser +from email.message import Message +from http import HTTPStatus +from http.server import HTTPServer, BaseHTTPRequestHandler +from urllib.parse import parse_qs, urlparse, urljoin, urlencode, uses_relative, uses_netloc +from contextlib import contextmanager + +# for urljoin: +uses_relative.append("gemini") +uses_netloc.append("gemini") + + +def _build_navigation(url=None): + form = ET.Element("form") + form.attrib.update(method="get") + input_ = ET.SubElement(form, "input") + input_.attrib.update( + **{ + "title": "url", + "type": "text", + "name": "url", + "placeholder": "gemini://", + "autocomplete": "off", + "size": "64", + } + ) + if url: + input_.attrib.update(value=url) + input_ = ET.SubElement(form, "input") + input_.attrib.update(**{"type": "submit", "value": "go!"}) + return ET.tostring(form) + b"\r\n" + + +class _HTTPServer(HTTPServer): + def __init__( + self, + *args, + header_file_path=None, + footer_file_path=None, + icon_file_path=None, + css_file_path=None, + **kwargs + ): + super().__init__(*args, **kwargs) + with open(header_file_path, "rb") as f: + self.header_file_bytes = f.read() + with open(footer_file_path, "rb") as f: + self.footer_file_bytes = f.read() + with open(icon_file_path, "rb") as f: + self.icon_file_bytes = f.read() + with open(css_file_path, "rb") as f: + self.css_file_bytes = f.read() + + +class _Elem: + def __init__(self, file): + self.elem = None + self._file = file + + @contextmanager + def __call__(self): + yield self + self.flush() + + def flush(self): + if self.elem is not None: + self._file.write(ET.tostring(self.elem) + b"\r\n") + self.elem = None + + +class _AutoFlush: + def __init__(self, elem): + self._elem = elem + + def cancel(self): + self._elem = None + + @contextmanager + def __call__(self): + yield self + if self._elem is not None: + self._elem.flush() + + +class _RequestHandler(BaseHTTPRequestHandler): + def _parse_path(self): + _, _, path, _, query, _ = urlparse(self.path) + return path, parse_qs(query) if query else {} + + def do_GET(self): + path, query = self._parse_path() + if path in {"/index.html", "/index.htm", "/index", "/"}: + url = query.get("url", [None])[0] + if not url: + self.send_response(HTTPStatus.OK) + self.send_header("Content-type", "text/html") + self.end_headers() + self.wfile.write( + self.server.header_file_bytes.replace(b"$URL", b"yet another http-to-gemini") + ) + self.wfile.write(_build_navigation()) + self.wfile.write(self.server.footer_file_bytes) + return + + try: + for _ in range(32): + parsed = urlparse(url) + if parsed.scheme != "gemini": + self.send_error(HTTPStatus.BAD_REQUEST, "Only gemini:// URLs are supported") + return + + context = ssl.SSLContext() + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE + with socket.create_connection((parsed.hostname, parsed.port or 1965)) as raw_s: + with context.wrap_socket(raw_s) as s: + s.sendall((url + '\r\n').encode("UTF-8")) + fp = s.makefile("rb") + splitted = fp.readline().decode("UTF-8").strip().split(maxsplit=1) + status = splitted[0] + if status.startswith("3") and len(splitted) == 2: + # redirect + url = urljoin(url, splitted[1]) + continue + if not status.startswith("2"): + self.send_error(HTTPStatus.INTERNAL_SERVER_ERROR, "Unsupported answer: " + str(splitted)) + return + mime = splitted[1].lower() if len(splitted) == 2 else "text/gemini" + if not mime.startswith("text/gemini"): + # return as-is + self.send_response(HTTPStatus.OK) + self.send_header("Content-type", mime) + self.end_headers() + self.wfile.write(fp.read()) + return + m = Message() + m['content-type'] = mime + body = fp.read().decode(m.get_param('charset') or "UTF-8") + self._convert_gemini_to_html(url, body, mime) + return + break + else: + raise RuntimeError("Too many redirects") + except Exception as error: + self.send_error(HTTPStatus.INTERNAL_SERVER_ERROR, str(error)) + return + + if path == "/favicon.ico": + self.send_response(HTTPStatus.OK) + self.send_header("Content-type", "image/x-icon") + self.end_headers() + self.wfile.write(self.server.icon_file_bytes) + return + + if path == "/style.css": + self.send_response(HTTPStatus.OK) + self.send_header("Content-type", "text/css") + self.end_headers() + self.wfile.write(self.server.css_file_bytes) + return + + self.send_error(HTTPStatus.NOT_FOUND, "File not found") + + def _convert_gemini_to_html(self, url, body, mime): + # convert gemini (body) to html + self.send_response(HTTPStatus.OK) + self.send_header("Content-type", mime.replace("gemini", "html")) + self.end_headers() + self.wfile.write(self.server.header_file_bytes.replace(b"$URL", url.encode())) + self.wfile.write(_build_navigation(url)) + + with _Elem(self.wfile)() as pre: + with _Elem(self.wfile)() as ul: + for line in body.splitlines(): + with _AutoFlush(ul)() as auto_flush_ul: + if line.startswith("```"): + if pre.elem is None: + pre.elem = ET.Element("pre") + pre.elem.text = "" + else: + pre.flush() + elif pre.elem is not None: + if pre.elem.text: + pre.elem.text += "\r\n" + pre.elem.text += line + elif line.startswith("=>") and line[2:].strip(): + p = ET.Element("p") + p.text = "=> " + splitted = line[2:].strip().split(maxsplit=1) + target = urljoin(url, splitted[0]) + a = ET.SubElement(p, "a") + if urlparse(target).scheme == "gemini": + a.attrib.update(href="/?" + urlencode({"url": target})) + else: + a.attrib.update(target="_blank", href=target) + a.text = splitted[1] if len(splitted) > 1 else target + self.wfile.write(ET.tostring(p) + b"\r\n") + elif line.startswith("#") and len(line.split()) > 1: + splitted = line.split(maxsplit=1) + h = ET.Element("h" + str(len(splitted[0]))) + h.text = splitted[1] + self.wfile.write(ET.tostring(h) + b"\r\n") + elif line.startswith("* ") and line[2:].strip(): + if ul.elem is None: + ul.elem = ET.Element("ul") + ET.SubElement(ul.elem, "li").text = line[2:].strip() + auto_flush_ul.cancel() + elif line.startswith("> ") and line[2:].strip(): + blockquote = ET.Element("blockquote") + ET.SubElement(blockquote, "p").text = line[2:].strip() + self.wfile.write(ET.tostring(blockquote) + b"\r\n") + else: + if line: + p = ET.Element("p") + p.text = line + self.wfile.write(ET.tostring(p) + b"\r\n") + self.wfile.write(self.server.footer_file_bytes) + + +def _main(): + parser = ArgumentParser() + parser.add_argument("--address", default="127.0.0.1", help="bind to this address (default: %(default)s)") + parser.add_argument("--port", default=8000, type=int, help="bind to this port (default: %(default)s)") + parser.add_argument("--header", required=True, help="path to `index.html.header`") + parser.add_argument("--footer", required=True, help="path to `index.html.footer`") + parser.add_argument("--icon", required=True , help="path to `favicon.ico`") + parser.add_argument("--css", required=True, help="path to `style.css`") + args = parser.parse_args() + + with _HTTPServer( + (args.address, args.port), + _RequestHandler, + header_file_path=args.header, + footer_file_path=args.footer, + icon_file_path=args.icon, + css_file_path=args.css, + ) as http_server: + sock_host, sock_port = http_server.socket.getsockname()[:2] + print(f"HTTP server started ({sock_host}:{sock_port})...") + try: + http_server.serve_forever() + except KeyboardInterrupt: + print("\nKeyboard interrupt received, exiting.") + + +if __name__ == '__main__': + _main()