1 """Yet another http-to-gemini."""
2 import socket
3 import ssl
4 import xml.etree.ElementTree as ET
5 from argparse import ArgumentParser
6 from email.message import Message
7 from http import HTTPStatus
8 from http.server import HTTPServer, BaseHTTPRequestHandler
9 from urllib.parse import parse_qs, urlparse, urljoin, urlencode, uses_relative, uses_netloc
10 from contextlib import contextmanager
11 
12 # for urljoin:
13 uses_relative.append("gemini")
14 uses_netloc.append("gemini")
15 
16 
17 def _build_navigation(url=None):
18     form = ET.Element("form")
19     form.attrib.update(method="get")
20     input_ = ET.SubElement(form, "input")
21     input_.attrib.update(
22         **{
23             "title": "url",
24             "type": "text",
25             "name": "url",
26             "placeholder": "gemini://",
27             "autocomplete": "off",
28             "size": "64",
29         }
30     )
31     if url:
32         input_.attrib.update(value=url)
33     input_ = ET.SubElement(form, "input")
34     input_.attrib.update(**{"type": "submit", "value": "go!"})
35     return ET.tostring(form) + b"\r\n"
36 
37 
38 class _HTTPServer(HTTPServer):
39     def __init__(
40         self,
41         *args,
42         header_file_path=None,
43         footer_file_path=None,
44         icon_file_path=None,
45         css_file_path=None,
46         **kwargs
47     ):
48         super().__init__(*args, **kwargs)
49         with open(header_file_path, "rb") as f:
50             self.header_file_bytes = f.read()
51         with open(footer_file_path, "rb") as f:
52             self.footer_file_bytes = f.read()
53         with open(icon_file_path, "rb") as f:
54             self.icon_file_bytes = f.read()
55         with open(css_file_path, "rb") as f:
56             self.css_file_bytes = f.read()
57 
58 
59 class _Elem:
60     def __init__(self, file):
61         self.elem = None
62         self.file = file
63 
64     @contextmanager
65     def __call__(self):
66         yield self
67         self.flush()
68 
69     def flush_bytes(self):
70         if self.elem is None:
71             return b""
72 
73         rv = ET.tostring(self.elem) + b"\r\n"
74         self.elem = None
75         return rv
76 
77     def flush(self):
78         self.file.write(self.flush_bytes())
79 
80 
81 class _FlushBeforeWrite:
82     def __init__(self, elem):
83         self._elem = elem
84         self._file = elem.file
85 
86     def cancel(self):
87         self._elem = None
88 
89     def commit(self):
90         if self._elem is not None:
91             self._elem.flush()
92             self._elem = None
93         return self._file
94 
95     @contextmanager
96     def __call__(self):
97         yield self
98         self.commit()
99 
100 
101 class _RequestHandler(BaseHTTPRequestHandler):
102     def _parse_path(self):
103         _, _, path, _, query, _ = urlparse(self.path)
104         return path, parse_qs(query) if query else {}
105 
106     def do_GET(self):
107         path, query = self._parse_path()
108         if path in {"/index.html", "/index.htm", "/index", "/"}:
109             url = query.get("url", [None])[0]
110             if not url:
111                 self.send_response(HTTPStatus.OK)
112                 self.send_header("Content-type", "text/html")
113                 self.end_headers()
114                 self.wfile.write(
115                     self.server.header_file_bytes.replace(b"$URL", b"yet another http-to-gemini")
116                 )
117                 self.wfile.write(_build_navigation())
118                 self.wfile.write(self.server.footer_file_bytes)
119                 return
120 
121             try:
122                 for _ in range(6): # first request + 5 consecutive redirects
123                     parsed = urlparse(url)
124                     if parsed.scheme != "gemini":
125                         self.send_error(HTTPStatus.BAD_REQUEST, "Only gemini:// URLs are supported")
126                         return
127 
128                     context = ssl.SSLContext()
129                     context.check_hostname = False
130                     context.verify_mode = ssl.CERT_NONE
131                     with socket.create_connection((parsed.hostname, parsed.port or 1965)) as raw_s:
132                         with context.wrap_socket(raw_s, server_hostname=parsed.hostname) as s:
133                             s.sendall((url + '\r\n').encode("UTF-8"))
134                             fp = s.makefile("rb")
135                             splitted = fp.readline().decode("UTF-8").strip().split(maxsplit=1)
136                             status = splitted[0]
137                             if status.startswith("3") and len(splitted) == 2:
138                                 # redirect
139                                 url = urljoin(url, splitted[1])
140                                 continue
141                             if not status.startswith("2"):
142                                 self.send_error(
143                                     HTTPStatus.INTERNAL_SERVER_ERROR,
144                                     f"Unsupported answer: {splitted[0]}",
145                                     splitted[1] if len(splitted) == 2 else None
146                                 )
147                                 return
148                             mime = splitted[1].lower() if len(splitted) == 2 else "text/gemini"
149                             if not mime.startswith("text/gemini"):
150                                 # return as-is
151                                 self.send_response(HTTPStatus.OK)
152                                 self.send_header("Content-type", mime)
153                                 self.end_headers()
154                                 self.wfile.write(fp.read())
155                                 return
156                             m = Message()
157                             m['content-type'] = mime
158                             body = fp.read().decode(m.get_param('charset') or "UTF-8")
159                             self._convert_gemini_to_html(url, body, mime)
160                             return
161                     break
162                 else:
163                     raise RuntimeError("Too many redirects")
164             except Exception as error:
165                 self.send_error(HTTPStatus.INTERNAL_SERVER_ERROR, str(error))
166                 return
167 
168         if path == "/favicon.ico":
169             self.send_response(HTTPStatus.OK)
170             self.send_header("Content-type", "image/x-icon")
171             self.end_headers()
172             self.wfile.write(self.server.icon_file_bytes)
173             return
174 
175         if path == "/style.css":
176             self.send_response(HTTPStatus.OK)
177             self.send_header("Content-type", "text/css")
178             self.end_headers()
179             self.wfile.write(self.server.css_file_bytes)
180             return
181 
182         self.send_error(HTTPStatus.NOT_FOUND, "File not found")
183 
184     def _convert_gemini_to_html(self, url, body, mime):
185         # convert gemini (body) to html
186         self.send_response(HTTPStatus.OK)
187         self.send_header("Content-type", mime.replace("gemini", "html"))
188         self.end_headers()
189         self.wfile.write(self.server.header_file_bytes.replace(b"$URL", url.encode()))
190         self.wfile.write(_build_navigation(url))
191         with _Elem(self.wfile)() as pre:
192             with _Elem(self.wfile)() as ul:
193                 for line in body.splitlines():
194                     with _FlushBeforeWrite(ul)() as flush_before:
195                         if line.startswith("```"):
196                             if pre.elem is None:
197                                 pre.elem = ET.Element("pre")
198                                 pre.elem.text = ""
199                             else:
200                                 flush_before.commit().write(pre.flush_bytes())
201                         elif pre.elem is not None:
202                             if pre.elem.text:
203                                 pre.elem.text += "\r\n"
204                             pre.elem.text += line
205                         elif line.startswith("=>") and line[2:].strip():
206                             p = ET.Element("p")
207                             p.text = "=> "
208                             splitted = line[2:].strip().split(maxsplit=1)
209                             target = urljoin(url, splitted[0])
210                             a = ET.SubElement(p, "a")
211                             if urlparse(target).scheme == "gemini":
212                                 a.attrib.update(href="/?" + urlencode({"url": target}))
213                             else:
214                                 a.attrib.update(target="_blank", href=target)
215                             a.text = splitted[1] if len(splitted) > 1 else target
216                             flush_before.commit().write(ET.tostring(p) + b"\r\n")
217                         elif line.startswith("#") and len(line.split()) > 1:
218                             splitted = line.split(maxsplit=1)
219                             h = ET.Element("h" + str(len(splitted[0])))
220                             h.text = splitted[1]
221                             flush_before.commit().write(ET.tostring(h) + b"\r\n")
222                         elif line.startswith("* ") and line[2:].strip():
223                             if ul.elem is None:
224                                 ul.elem = ET.Element("ul")
225                             ET.SubElement(ul.elem, "li").text = line[2:].strip()
226                             flush_before.cancel()
227                         elif line.startswith("> ") and line[2:].strip():
228                             blockquote = ET.Element("blockquote")
229                             ET.SubElement(blockquote, "p").text = line[2:].strip()
230                             flush_before.commit().write(ET.tostring(blockquote) + b"\r\n")
231                         else:
232                             if line:
233                                 p = ET.Element("p")
234                                 p.text = line
235                                 flush_before.commit().write(ET.tostring(p) + b"\r\n")
236         self.wfile.write(self.server.footer_file_bytes)
237 
238 
239 def _main():
240     parser = ArgumentParser()
241     parser.add_argument("--address", default="127.0.0.1", help="bind to this address (default: %(default)s)")
242     parser.add_argument("--port", default=8000, type=int, help="bind to this port (default: %(default)s)")
243     parser.add_argument("--header", required=True, help="path to `index.html.header`")
244     parser.add_argument("--footer", required=True, help="path to `index.html.footer`")
245     parser.add_argument("--icon", required=True , help="path to `favicon.ico`")
246     parser.add_argument("--css", required=True, help="path to `style.css`")
247     args = parser.parse_args()
248 
249     with _HTTPServer(
250         (args.address, args.port),
251         _RequestHandler,
252         header_file_path=args.header,
253         footer_file_path=args.footer,
254         icon_file_path=args.icon,
255         css_file_path=args.css,
256     ) as http_server:
257         sock_host, sock_port = http_server.socket.getsockname()[:2]
258         print(f"HTTP server started ({sock_host}:{sock_port})...")
259         try:
260             http_server.serve_forever()
261         except KeyboardInterrupt:
262             print("\nKeyboard interrupt received, exiting.")
263 
264 
265 if __name__ == '__main__':
266     _main()