]>
git.mar77i.info Git - hublib/blob - hub/hubapp.py
591856ce81c86bfb68977abe527d6fd4a325b653
1 from base64
import urlsafe_b64encode
2 from hashlib
import pbkdf2_hmac
3 from pathlib
import Path
4 from secrets
import token_urlsafe
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
11 from .websocket
import WebSocketApp
13 MEDIA_CSS
= "text/css"
18 Basic static file wrapper.
20 media_types_per_suffix
= {
28 def __init__(self
, path
):
31 self
.media_type
= self
.get_media_type(path
)
36 def get_media_type(cls
, path
):
37 return cls
.media_types_per_suffix
[path
.suffix
]
40 mtime
= self
.path
.stat().st_mtime
41 if mtime
!= self
.mtime
:
42 with open(self
.path
) as fh
:
43 self
.content
= fh
.read()
48 return f
"<{type(self).__name__} {self.path}>"
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
59 def is_ignored_filename(path
):
60 return path
.name
.startswith(".")
63 def scan_files(cls
, base_dir
):
64 stack
= [base_dir
.iterdir()]
67 path
= next(stack
[-1])
72 stack
.append(path
.iterdir())
73 elif not cls
.is_ignored_filename(path
):
76 def get_file(self
, path
: Path
) -> StaticFile
:
77 return StaticFile(path
)
79 def uri_tail(self
, path
: Path
) -> str:
81 Return the "local" path, relative to self.base_dir, if applicable
83 assert isinstance(path
, Path
)
84 if path
.is_absolute():
85 path
= path
.relative_to(self
.base_dir
)
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('/')}"
97 def __init__(self
, app
, base_dir
, base_uri
="/"):
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
)
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
116 class StaticTemplateFile(StaticFile
):
117 TEMPLATE_SUFFIX
= ".j2"
118 content
: Template |
None
120 def __init__(self
, path
, hubapp
):
121 super().__init
__(path
)
122 self
.name
= path
.stem
124 self
.context
= {"static_file": self}
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]))
134 return self
.content
.render(self
.context
)
137 def get_media_type(cls
, path
):
138 assert path
.suffix
== cls
.TEMPLATE_SUFFIX
139 return cls
.media_types_per_suffix
[path
.suffixes
[-2]]
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
)
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
)]
155 class RootApp(TemplateTreeFileApp
):
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
):
162 def __init__(self
, app
, base_dir
, base_uri
="/", secret
=None):
163 from .utils
import get_redis_pass
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"))
169 async def setup(self
):
170 await self
.conn
.set("client_id", 0)
172 def scramble(self
, value
):
173 if isinstance(value
, str):
174 value
= value
.encode()
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")
182 def uri_tail(self
, path
: Path
) -> str:
183 return self
.scramble(super().uri_tail(path
))
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
)
191 self
.uri(Path(f
"ws_{value}")): value
for value
in ("client", "master")
193 for uri
, suffix
in self
.ws_uris
.items():
194 app
.add_route(uri
, self
.ws_app
, suffix
=suffix
)
197 def is_master_uri(path
: Path
) -> bool:
198 assert isinstance(path
, Path
)
200 pos
= basename
.find(".")
202 basename
= basename
[:pos
]
203 return basename
in "master"
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
)
213 uri
= f
"{uri[:pos]}/{scrambled_uri}"