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

   1 from base64 import urlsafe_b64encode
   2 from hashlib import pbkdf2_hmac
   3 from pathlib import Path
   4 from secrets import token_urlsafe
   5
   6 from falcon.constants import MEDIA_HTML, MEDIA_JS, MEDIA_TEXT
   7 from falcon.status_codes import HTTP_OK
   8 from jinja2 import Template
   9 from redis.asyncio import StrictRedis
  10
  11 from .websocket import WebSocketApp
  12
  13 MEDIA_CSS = "text/css"
  14
  15
  16 class StaticFile:
  17     """
  18     Basic static file wrapper.
  19     """
  20     media_types_per_suffix = {
  21         ".html": MEDIA_HTML,
  22         ".js": MEDIA_JS,
  23         ".css": MEDIA_CSS,
  24         ".txt": MEDIA_TEXT,
  25     }
  26     content: str | None
  27
  28     def __init__(self, path):
  29         self.path = path
  30         self.name = path.name
  31         self.media_type = self.get_media_type(path)
  32         self.mtime = None
  33         self.content = None
  34
  35     @classmethod
  36     def get_media_type(cls, path):
  37         return cls.media_types_per_suffix[path.suffix]
  38
  39     def get(self):
  40         mtime = self.path.stat().st_mtime
  41         if mtime != self.mtime:
  42             with open(self.path) as fh:
  43                 self.content = fh.read()
  44             self.mtime = mtime
  45         return self.content
  46
  47     def __repr__(self):
  48         return f"<{type(self).__name__} {self.path}>"
  49
  50
  51 class TreeFileApp:
  52     """
  53     Map a directory tree base_dir to a base_uri and serve it statically.
  54     Map index.html files to their relative roots:
  55     index.html to "/" (root uri) to index.html and
  56     "/a" (directory uri) to a/index.html
  57     """
  58     @staticmethod
  59     def is_ignored_filename(path):
  60         return path.name.startswith(".")
  61
  62     @classmethod
  63     def scan_files(cls, base_dir):
  64         stack = [base_dir.iterdir()]
  65         while len(stack):
  66             try:
  67                 path = next(stack[-1])
  68             except StopIteration:
  69                 stack.pop()
  70                 continue
  71             if path.is_dir():
  72                 stack.append(path.iterdir())
  73             elif not cls.is_ignored_filename(path):
  74                 yield path
  75
  76     def get_file(self, path: Path) -> StaticFile:
  77         return StaticFile(path)
  78
  79     def uri_tail(self, path: Path) -> str:
  80         """
  81         Return the "local" path, relative to self.base_dir, if applicable
  82         """
  83         assert isinstance(path, Path)
  84         if path.is_absolute():
  85             path = path.relative_to(self.base_dir)
  86         return str(path)
  87
  88     def uri(self, path: Path) -> str:
  89         uri_tail = self.uri_tail(path)
  90         if uri_tail == "index.html":
  91             return self.base_uri or "/"
  92         index_suffix = "/index.html"
  93         if uri_tail.endswith(index_suffix):
  94             uri_tail = uri_tail[:-len(index_suffix)]
  95         return f"{self.base_uri}/{uri_tail.lstrip('/')}"
  96
  97     def __init__(self, app, base_dir, base_uri="/"):
  98         self.app = app
  99         self.base_dir = base_dir
 100         self.base_uri = base_uri.rstrip("/")
 101         self.name = self.base_uri.replace("/", ".").strip(".") or "root"
 102         self.files_per_uris = {}
 103         for path in self.scan_files(base_dir):
 104             static_file = self.get_file(path)
 105             uri = self.uri(static_file.path)
 106             self.files_per_uris[uri] = static_file
 107             app.add_route(uri, self)
 108
 109     async def on_get(self, req, resp):
 110         resource = self.files_per_uris[req.path]
 111         resp.content_type = resource.media_type
 112         resp.data = resource.get().encode()
 113         resp.status = HTTP_OK
 114
 115
 116 class StaticTemplateFile(StaticFile):
 117     TEMPLATE_SUFFIX = ".j2"
 118     content: Template | None
 119
 120     def __init__(self, path, hubapp):
 121         super().__init__(path)
 122         self.name = path.stem
 123         self.hubapp = hubapp
 124         self.context = {"static_file": self}
 125
 126     def get(self) -> str:
 127         mtime = self.path.stat().st_mtime
 128         if mtime != self.mtime:
 129             env = self.hubapp.app.env
 130             self.content = env.get_template(
 131                 str(self.path.relative_to(env.loader.searchpath[0]))
 132             )
 133             self.mtime = mtime
 134         return self.content.render(self.context)
 135
 136     @classmethod
 137     def get_media_type(cls, path):
 138         assert path.suffix == cls.TEMPLATE_SUFFIX
 139         return cls.media_types_per_suffix[path.suffixes[-2]]
 140
 141
 142 class TemplateTreeFileApp(TreeFileApp):
 143     def get_file(self, path: Path) -> StaticFile:
 144         if path.suffix == StaticTemplateFile.TEMPLATE_SUFFIX:
 145             return StaticTemplateFile(path, self)
 146         return super().get_file(path)
 147
 148     def uri_tail(self, path: Path) -> str:
 149         uri_tail = super().uri_tail(path)
 150         if uri_tail.endswith(StaticTemplateFile.TEMPLATE_SUFFIX):
 151             uri_tail = uri_tail[:-len(StaticTemplateFile.TEMPLATE_SUFFIX)]
 152         return uri_tail
 153
 154
 155 class RootApp(TemplateTreeFileApp):
 156     @classmethod
 157     def scan_files(cls, base_dir):
 158         for path in base_dir.iterdir():
 159             if not path.is_dir() and not cls.is_ignored_filename(path):
 160                 yield path
 161
 162     def __init__(self, app, base_dir, base_uri="/", secret=None):
 163         from .utils import get_redis_pass
 164
 165         self.secret = secret or token_urlsafe(64)
 166         super().__init__(app, base_dir, base_uri)
 167         self.conn = StrictRedis(username="default", password=get_redis_pass("/etc/redis/redis.conf"))
 168
 169     async def setup(self):
 170         await self.conn.set("client_id", 0)
 171
 172     def scramble(self, value):
 173         if isinstance(value, str):
 174             value = value.encode()
 175         secret = self.secret
 176         if isinstance(secret, str):
 177             secret = secret.encode()
 178         return urlsafe_b64encode(
 179             pbkdf2_hmac("sha512", value, secret, 221100)
 180         ).rstrip(b"=").decode("ascii")
 181
 182     def uri_tail(self, path: Path) -> str:
 183         return self.scramble(super().uri_tail(path))
 184
 185
 186 class HubApp(TemplateTreeFileApp):
 187     def __init__(self, app, base_dir, base_uri="/"):
 188         super().__init__(app, base_dir, base_uri)
 189         self.ws_app = WebSocketApp(self)
 190         self.ws_uris = {
 191             self.uri(Path(f"ws_{value}")): value for value in ("client", "master")
 192         }
 193         for uri, suffix in self.ws_uris.items():
 194             app.add_route(uri, self.ws_app, suffix=suffix)
 195
 196     @staticmethod
 197     def is_master_uri(path: Path) -> bool:
 198         assert isinstance(path, Path)
 199         basename = path.name
 200         pos = basename.find(".")
 201         if pos != -1:
 202             basename = basename[:pos]
 203         return basename in "master"
 204
 205     def uri(self, path: Path) -> str:
 206         uri = super().uri(path)
 207         if self.is_master_uri(path):
 208             pos = uri.rstrip("/").rfind("/")
 209             scrambled_uri = self.app.hubapps["root"].scramble(uri)
 210             if pos == -1:
 211                 uri = scrambled_uri
 212             else:
 213                 uri = f"{uri[:pos]}/{scrambled_uri}"
 214         return uri