git.mar77i.info Git - hublib/blob - hub/app.py

   1 import socket
   2 import sys
   3 from argparse import ArgumentParser
   4 from base64 import urlsafe_b64encode
   5 from hashlib import pbkdf2_hmac
   6 from itertools import chain
   7 from pathlib import Path
   8 from secrets import token_urlsafe
   9 from subprocess import run
  10 from traceback import print_exception
  11 from urllib.parse import urlunsplit
  12
  13 from falcon.asgi import App as FalconApp
  14 from falcon.constants import MEDIA_HTML
  15 from jinja2 import Environment, FileSystemLoader, select_autoescape
  16 from uvicorn import Config, Server
  17
  18 from .hubapp import HubApp, RootApp
  19
  20
  21 class App(FalconApp):
  22     def __init__(self, secret, **kwargs):
  23         kwargs.setdefault("media_type", MEDIA_HTML)
  24         super().__init__(**kwargs)
  25         self.base_dir = Path(__file__).parents[1] / "webroot"
  26         self.env = Environment(
  27             loader=FileSystemLoader(self.base_dir),
  28             autoescape=select_autoescape(),
  29             extensions=["hub.utils.StaticTag"],
  30         )
  31         self.secret = secret or token_urlsafe(64)
  32         self.hubapps = {}
  33         RootApp(self, self.base_dir)
  34         for base_dir in self.base_dir.iterdir():
  35             if not base_dir.is_dir() or HubApp.is_ignored_filename(base_dir):
  36                 continue
  37             HubApp(self, base_dir)
  38         self.add_error_handler(Exception, self.print_exception)
  39
  40     def scramble(self, value):
  41         if isinstance(value, str):
  42             value = value.encode()
  43         secret = self.secret
  44         if isinstance(secret, str):
  45             secret = secret.encode()
  46         return urlsafe_b64encode(
  47             pbkdf2_hmac("sha512", value, secret, 221100)
  48         ).rstrip(b"=").decode("ascii")
  49
  50     async def print_exception(self, req, resp, ex, params):
  51         print_exception(*sys.exc_info())
  52
  53
  54 class HubServer(Server):
  55     def __init__(self, *args, browser, **kwargs):
  56         self.browser = browser
  57         super().__init__(*args, **kwargs)
  58
  59     async def startup(self, sockets: list[socket.socket] | None = None) -> None:
  60         await super().startup(sockets)
  61         config = self.config
  62         protocol_name = "https" if config.ssl else "http"
  63         host = "0.0.0.0" if config.host is None else config.host
  64         if ":" in host:
  65             # It's an IPv6 address.
  66             host = f"[{host.rstrip(']').lstrip('[')}]"
  67
  68         port = config.port
  69         if port == 0:
  70             try:
  71                 port = next(
  72                     chain.from_iterable(
  73                         (server.sockets for server in getattr(self, "servers", ()))
  74                     )
  75                 ).getsockname()[1]
  76             except StopIteration:
  77                 pass
  78         if {"http": 80, "https": 443}[protocol_name] != port:
  79             host = f"{host}:{port}"
  80         app = config.loaded_app.app
  81         print("Secret:", app.secret)
  82         for key, value in app.hubapps["root"].files_per_uris.items():
  83             if Path(value.path.name).stem == "index.html":
  84                 url = urlunsplit((protocol_name, host, key, "", ""))
  85                 print("URL:", url)
  86                 if self.browser:
  87                     run([self.browser, url])
  88
  89
  90 app: App
  91
  92
  93 async def main():
  94     global app
  95     ap = ArgumentParser()
  96     ap.add_argument("--secret")
  97     ap.add_argument("--browser", default="firefox")
  98     args = ap.parse_args()
  99     app = App(args.secret)
 100     config = Config("hub.app:app", port=5000, log_level="info")
 101     config.setup_event_loop()
 102     hs = HubServer(config, browser=args.browser)
 103     await hs.serve()