]>
git.mar77i.info Git - hublib/blob - hub/app.py
3 from argparse
import ArgumentParser
4 from base64
import urlsafe_b64encode
5 from hashlib
import pbkdf2_hmac
6 from pathlib
import Path
7 from secrets
import token_urlsafe
8 from subprocess
import run
9 from traceback
import print_exception
10 from urllib
.parse
import urlunsplit
12 from falcon
.asgi
import App
as FalconApp
13 from falcon
.constants
import MEDIA_HTML
14 from jinja2
import Environment
, FileSystemLoader
, select_autoescape
15 from redis
.asyncio
import StrictRedis
16 from uvicorn
import Config
, Server
18 from .static
import HubApp
, TemplateTreeFileApp
21 class App(TemplateTreeFileApp
, FalconApp
):
23 def scan_files(cls
, base_dir
):
24 for path
in base_dir
.iterdir():
25 if not path
.is_dir() and not cls
.is_ignored_filename(path
):
28 def scramble(self
, value
):
29 if isinstance(value
, str):
30 value
= value
.encode()
32 if isinstance(secret
, str):
33 secret
= secret
.encode()
34 return urlsafe_b64encode(
35 pbkdf2_hmac("sha512", value
, secret
, 221100)
36 ).rstrip(b
"=").decode("ascii")
38 def uri_tail(self
, path
: Path
) -> str:
39 return self
.scramble(super().uri_tail(path
))
41 def __init__(self
, base_dir
, secret
, **kwargs
):
42 from .utils
import get_redis_pass
44 self
.secret
= secret
or token_urlsafe(64)
45 kwargs
.setdefault("media_type", MEDIA_HTML
)
46 FalconApp
.__init
__(self
, **kwargs
)
47 TemplateTreeFileApp
.__init
__(
48 self
, self
, base_dir
, "/derp", "root"
50 self
.env
= Environment(
51 loader
=FileSystemLoader(self
.base_dir
),
52 autoescape
=select_autoescape(),
53 extensions
=["hub.utils.StaticTag"],
56 self
.conn
= StrictRedis(
57 username
="default", password
=get_redis_pass("/etc/redis/redis.conf")
59 for base_dir
in self
.base_dir
.iterdir():
60 if base_dir
.is_dir() and not self
.is_ignored_filename(base_dir
):
61 self
.hubapps
[base_dir
.name
] = HubApp(
62 self
, base_dir
, base_uri
=f
"/derp/{base_dir.name}"
64 self
.add_error_handler(Exception, self
.print_exception
)
66 async def print_exception(self
, req
, resp
, ex
, params
):
67 print_exception(*sys
.exc_info())
70 class HubServer(Server
):
71 def __init__(self
, *args
, browser
, **kwargs
):
72 self
.browser
= browser
73 super().__init
__(*args
, **kwargs
)
75 async def startup(self
, sockets
: list[socket
.socket
] |
None = None) -> None:
76 await super().startup(sockets
)
77 root_app
= self
.config
.loaded_app
.app
78 print("Secret:", root_app
.secret
)
79 for uri
, file in root_app
.files_per_uris
.items():
80 if file.name
== "index.html":
83 raise ValueError("Root page not found!")
84 host
, port
, ssl
= self
.config
.host
, self
.config
.port
, bool(self
.config
.ssl
)
85 if port
and port
!= (80, 443)[ssl
]:
86 host
= f
"{host}:{port}"
87 url
= urlunsplit((f
"http{'s'[:ssl]}", host
, uri
, "", ""))
90 run([self
.browser
, url
])
95 ap
.add_argument("--secret")
96 ap
.add_argument("--browser", default
="xdg-open")
97 args
= ap
.parse_args()
98 app
= App(Path(__file__
).parents
[1] / "webroot", args
.secret
)
99 await app
.conn
.set("client_id", 0)
100 config
= Config(app
, port
=5000, log_level
="info")
101 config
.setup_event_loop()
102 hs
= HubServer(config
, browser
=args
.browser
)