oni.git

oni

~ A static Git browser written in Rust.
git clone https://github.com/debuman/oni.git

scripts/serve.py (5564B)

      1 #!/usr/bin/env python3
      2 """Serve only the generated static site on the local loopback interface."""
      3 
      4 import argparse
      5 from functools import partial
      6 from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
      7 import os
      8 from pathlib import Path
      9 from threading import BoundedSemaphore
     10 from urllib.parse import unquote, urlsplit
     11 
     12 PUBLIC = Path(__file__).resolve().parents[1] / "public"
     13 CSP = (Path(__file__).resolve().parents[1] / "security/csp.txt").read_text().strip()
     14 is_reserved_path = getattr(os.path, "isreserved", lambda value: os.name == "nt" and Path(value).is_reserved())
     15 
     16 
     17 class PreviewServer(ThreadingHTTPServer):
     18     daemon_threads = True
     19 
     20     def __init__(self, *args, **kwargs):
     21         self.slots = BoundedSemaphore(32)
     22         super().__init__(*args, **kwargs)
     23 
     24     def process_request(self, request, client_address):
     25         if not self.slots.acquire(blocking=False):
     26             self.shutdown_request(request)
     27             return
     28         try:
     29             super().process_request(request, client_address)
     30         except BaseException:
     31             self.slots.release()
     32             raise
     33 
     34     def process_request_thread(self, request, client_address):
     35         try:
     36             super().process_request_thread(request, client_address)
     37         finally:
     38             self.slots.release()
     39 
     40 
     41 class PreviewHandler(SimpleHTTPRequestHandler):
     42     def setup(self):
     43         super().setup()
     44         self.connection.settimeout(10)
     45 
     46     def end_headers(self):
     47         self.send_header("Cache-Control", "no-store")
     48         self.send_header("Content-Security-Policy", CSP + "; frame-ancestors 'none'")
     49         self.send_header("X-Content-Type-Options", "nosniff")
     50         self.send_header("X-Frame-Options", "DENY")
     51         self.send_header("Referrer-Policy", "no-referrer")
     52         self.send_header("Cross-Origin-Resource-Policy", "same-origin")
     53         super().end_headers()
     54 
     55     def send_head(self):
     56         # Binding to loopback alone does not prevent DNS rebinding.
     57         hosts = self.headers.get_all("Host", [])
     58         try:
     59             authority = urlsplit("//" + hosts[0]) if len(hosts) == 1 else None
     60             valid_host = (authority is not None and authority.hostname in ("localhost", "127.0.0.1", "::1")
     61                           and authority.port in (None, self.server.server_port)
     62                           and authority.username is None and authority.password is None
     63                           and not authority.path and not authority.query and not authority.fragment)
     64         except ValueError:
     65             valid_host = False
     66         if not valid_host:
     67             self.send_error(403, "Invalid preview host")
     68             return None
     69         if self.headers.get("Sec-Fetch-Site") == "cross-site":
     70             self.send_error(403, "Cross-site requests are not allowed")
     71             return None
     72         origin = self.headers.get("Origin")
     73         if origin and origin != "http://" + hosts[0]:
     74             self.send_error(403, "Invalid preview origin")
     75             return None
     76         try:
     77             url = urlsplit(self.path)
     78             decoded = unquote(url.path, errors="strict")
     79             parts = decoded.split("/")
     80             if (url.scheme or url.netloc or '\\' in decoded or ':' in decoded
     81                     or any(ord(char) < 32 or ord(char) == 127 for char in decoded)
     82                     or any(part and (part.endswith((".", " ")) or part.lower() == ".git"
     83                                      or is_reserved_path(part)) for part in parts)):
     84                 raise ValueError("Unsafe preview path")
     85             root = Path(self.directory).resolve()
     86             target = root
     87             for part in parts:
     88                 if not part:
     89                     continue
     90                 target = target / part
     91                 if target.is_symlink() or target.is_junction():
     92                     raise ValueError("Links are not served")
     93             target.resolve().relative_to(root)
     94             if target.is_dir():
     95                 for index in (target / "index.html", target / "index.htm"):
     96                     if index.is_symlink() or index.is_junction():
     97                         raise ValueError("Linked indexes are not served")
     98                     index.resolve().relative_to(root)
     99             elif target.exists() and not target.is_file():
    100                 raise ValueError("Only regular files are served")
    101         except (ValueError, OSError, UnicodeError):
    102             self.send_error(403, "Path is outside the preview")
    103             return None
    104         return super().send_head()
    105 
    106     def guess_type(self, path):
    107         if "raw" in (part.lower() for part in Path(path).relative_to(Path(self.directory)).parts):
    108             return {".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
    109                     ".gif": "image/gif", ".webp": "image/webp", ".ico": "image/vnd.microsoft.icon",
    110                     ".avif": "image/avif"}.get(Path(path).suffix.lower(), "application/octet-stream")
    111         return super().guess_type(path)
    112 
    113     def list_directory(self, path):
    114         self.send_error(404, "Page not found")
    115         return None
    116 
    117 
    118 if __name__ == "__main__":
    119     parser = argparse.ArgumentParser(description=__doc__)
    120     parser.add_argument("--port", type=int, default=4174)
    121     args = parser.parse_args()
    122     if not (PUBLIC / "index.html").is_file():
    123         parser.error("Run build.ps1 first")
    124     with PreviewServer(("127.0.0.1", args.port), partial(PreviewHandler, directory=str(PUBLIC))) as server:
    125         print(f"Local stagit preview: http://127.0.0.1:{args.port}/", flush=True)
    126         server.serve_forever()