"""librus-tui - a terminal-looking web frontend for Librus Synergia. Run: python app.py # real mode, http://127.0.0.1:8765 python app.py --demo # offline demo data The Synergia client is adapted from dani3l0/librusik (stable branch). """ import argparse import asyncio import json import logging import os import re import secrets import time from dataclasses import dataclass, field from pathlib import Path from urllib.parse import urlparse from aiohttp import web from bs4 import BeautifulSoup, Comment from demo import DemoLibrus from librus import Librus BASE_DIR = Path(__file__).resolve().parent STATIC_DIR = BASE_DIR / "static" COOKIE_NAME = "librustui_session" SESSION_IDLE = 12 * 60 * 60 SESSION_MAX = 24 * 60 * 60 VERSION = "1.0.0" log = logging.getLogger("librustui") # --------------------------------------------------------------------------- errors class ApiError(Exception): def __init__(self, status, message): super().__init__(message) self.status = status self.message = message # --------------------------------------------------------------------------- sessions @dataclass class Session: token: str client: object login: str demo: bool = False created: float = field(default_factory=time.time) last_seen: float = field(default_factory=time.time) def touch(self): self.last_seen = time.time() def expired(self): now = time.time() return ( now - self.last_seen > SESSION_IDLE or now - self.created > SESSION_MAX ) SESSIONS: dict[str, Session] = {} LOGIN_ATTEMPTS: dict[str, list[float]] = {} def cleanup_sessions(): for token in [t for t, s in SESSIONS.items() if s.expired()]: SESSIONS.pop(token, None) # --------------------------------------------------------------------------- html sanitizer ALLOWED_TAGS = { "p", "br", "b", "strong", "i", "em", "u", "s", "strike", "del", "ins", "ul", "ol", "li", "a", "img", "blockquote", "pre", "code", "hr", "h1", "h2", "h3", "h4", "h5", "h6", "div", "span", "font", "table", "thead", "tbody", "tfoot", "tr", "td", "th", "caption", "center", "small", "big", "sub", "sup", "dl", "dt", "dd", } ALLOWED_ATTRS = { "a": {"href", "title"}, "img": {"src", "alt", "title", "width", "height"}, "td": {"colspan", "rowspan"}, "th": {"colspan", "rowspan"}, "font": {"color", "size"}, } SAFE_URL = re.compile(r"^(https?://|mailto:|#|/)") def sanitize_html(html): """Allow a small set of formatting tags; drop scripts and handlers.""" if not html: return "" soup = BeautifulSoup(html, "html.parser") for comment in soup.find_all(string=lambda text: isinstance(text, Comment)): comment.extract() for tag in soup.find_all(True): if tag.name not in ALLOWED_TAGS: tag.unwrap() continue allowed = ALLOWED_ATTRS.get(tag.name, set()) attrs = {} for key, value in list(tag.attrs.items()): key_lower = key.lower() if key_lower not in allowed or key_lower.startswith("on"): continue if key_lower in ("href", "src"): value = str(value).strip() if not SAFE_URL.match(value): continue if value.startswith("/"): value = "https://synergia.librus.pl" + value attrs[key] = value tag.attrs = attrs if tag.name == "a" and attrs.get("href"): tag["target"] = "_blank" tag["rel"] = "noopener noreferrer" return str(soup) # --------------------------------------------------------------------------- helpers def safe_json(data): return json.dumps(data, ensure_ascii=False, default=str) def json_response(data, status=200): return web.Response( text=safe_json(data), status=status, content_type="application/json", ) def err(status, message): return json_response({"error": message}, status=status) def get_session(request): session = request.get("session") if session is None: raise ApiError(401, "Not authenticated.") return session def validate_message_link(link): """Allow Unicode attachment paths, but no absolute paths or traversal.""" if not link or len(link) > 500: return False if link.startswith(("/", "\\")): return False if any(ord(char) < 32 for char in link): return False if any(char in link for char in ('"', "'", "<", ">", "\\")): return False if any(part in (".", "..") for part in link.split("/")): return False return True async def safe_call(coro, default=None): """Await a client call, returning default instead of exploding.""" try: return await coro except Exception as exc: # upstream markup/JSON changes are not fatal log.warning("upstream call failed: %s", exc) return default # --------------------------------------------------------------------------- API async def api_config(request): return json_response( { "demo": bool(request.app["demo"]), "version": VERSION, } ) async def api_session(request): session = request.get("session") return json_response( { "authenticated": session is not None, "login": session.login if session else None, "demo": session.demo if session else False, "demoAvailable": bool(request.app["demo"]), } ) async def api_login(request): body = await request.json() if not isinstance(body, dict): raise ApiError(400, "Expected a JSON object.") cleanup_sessions() remote = request.remote or "unknown" now = time.time() attempts = [t for t in LOGIN_ATTEMPTS.get(remote, []) if now - t < 60] if len(attempts) >= 6: raise ApiError(429, "Too many login attempts. Wait a minute.") attempts.append(now) LOGIN_ATTEMPTS[remote] = attempts demo_requested = bool(body.get("demo")) if request.app["demo"] and demo_requested: client = DemoLibrus() token = secrets.token_urlsafe(32) SESSIONS[token] = Session(token=token, client=client, login="demo", demo=True) response = json_response({"ok": True, "demo": True}) response.set_cookie( COOKIE_NAME, token, max_age=SESSION_MAX, httponly=True, samesite="Lax", path="/", ) return response # A fresh login always replaces the previous server-side session. old_token = request.cookies.get(COOKIE_NAME) if old_token: SESSIONS.pop(old_token, None) login = body.get("login") password = body.get("password") if not isinstance(login, str) or not isinstance(password, str): raise ApiError(400, "Login and password are required.") login = login.strip() if not login or not password: raise ApiError(400, "Login and password are required.") if len(login) > 200 or len(password) > 500: raise ApiError(400, "Credentials are too long.") client = Librus() try: ok = await asyncio.wait_for(client.mktoken(login, password), timeout=55) except asyncio.TimeoutError: raise ApiError(504, "Synergia did not answer in time.") if not ok: if client.credentials_rejected: raise ApiError(401, "Librus rejected these credentials.") raise ApiError( 502, "Could not reach Synergia (outage, captcha or network error). " "Try again in a moment.", ) token = secrets.token_urlsafe(32) SESSIONS[token] = Session(token=token, client=client, login=login) response = json_response({"ok": True, "demo": False}) response.set_cookie( COOKIE_NAME, token, max_age=SESSION_MAX, httponly=True, samesite="Lax", path="/", ) log.info("login ok for %s", login) return response async def api_logout(request): session = request.get("session") if session: SESSIONS.pop(session.token, None) response = json_response({"ok": True}) response.del_cookie(COOKIE_NAME, path="/") return response async def api_bootstrap(request): session = get_session(request) client = session.client me, school, lucky, behaviour = await asyncio.gather( safe_call(client.get_me()), safe_call(client.get_school()), safe_call(client.get_lucky_number()), safe_call(client.get_behaviour_grade()), ) if me is None: SESSIONS.pop(session.token, None) raise ApiError( 502, "Synergia did not return account data. The session may have expired.", ) return json_response( { "me": me, "school": school, "luckyNumber": lucky, "behaviour": behaviour, } ) async def api_notifications(request): session = get_session(request) data = await safe_call(session.client.get_notifications()) if data is None: return json_response( {"grades": 0, "exams": 0, "absences": 0, "conferences": 0, "unavailable": True} ) return json_response(data) async def api_grades(request): session = get_session(request) data = await safe_call(session.client.get_grades()) if data is None: raise ApiError(502, "Could not load grades from Synergia.") return json_response({"subjects": data}) async def api_timetable(request): session = get_session(request) try: week = int(request.query.get("week", "0")) except ValueError: week = 0 if not 0 <= week <= 3: raise ApiError(400, "week must be between 0 and 3.") data = await safe_call(session.client.get_timetable(week)) if data is None: raise ApiError(502, "Could not load the timetable from Synergia.") return json_response(data) async def api_exams(request): session = get_session(request) data = await safe_call(session.client.get_exams()) if data is None: raise ApiError(502, "Could not load homework from Synergia.") return json_response({"exams": data}) async def api_attendances(request): session = get_session(request) data = await safe_call(session.client.get_attendances()) if data is None: raise ApiError(502, "Could not load attendance from Synergia.") return json_response({"attendances": data}) async def api_messages(request): session = get_session(request) data = await safe_call(session.client.get_messages(), default=[]) return json_response({"messages": data, "demo": session.demo}) async def api_message(request): session = get_session(request) link = request.query.get("link", "") if not validate_message_link(link): raise ApiError(400, "Invalid message link.") data = await safe_call(session.client.get_message(link)) if data is None: raise ApiError(502, "Could not load the message from Synergia.") data = dict(data) data["content"] = sanitize_html(data.get("content", "")) return json_response(data) async def api_download(request): session = get_session(request) link = request.query.get("link", "") if not validate_message_link(link): raise ApiError(400, "Invalid attachment link.") result = await safe_call(session.client.download_file(link)) if not result: raise ApiError(404, "Attachment not found or already expired.") headers = result.get("headers") or {} content_type = headers.get("Content-Type", "application/octet-stream") disposition = headers.get("Content-Disposition") or "" # Always force a download, even if Librus forgot the attachment header. if not disposition.lower().startswith("attachment"): name = link.rsplit("/", 1)[-1] or "attachment" name = re.sub(r"[^A-Za-z0-9._-]", "_", name) disposition = 'attachment; filename="%s"' % name response = web.Response(body=result.get("content", b"")) response.headers["Content-Type"] = content_type response.headers["Content-Disposition"] = disposition response.headers["X-Content-Type-Options"] = "nosniff" response.headers["Content-Security-Policy"] = "sandbox" return response async def api_freedays(request): session = get_session(request) school, teachers = await asyncio.gather( safe_call(session.client.get_free_days(), default=[]), safe_call(session.client.get_teacher_free_days(), default=[]), ) return json_response({"school": school or [], "teachers": teachers or []}) async def api_conferences(request): session = get_session(request) data = await safe_call(session.client.get_parent_teacher_conferences(), default=[]) return json_response({"conferences": data or []}) # --------------------------------------------------------------------------- static async def index(request): response = web.FileResponse(STATIC_DIR / "index.html") response.headers["Cache-Control"] = "no-store" return response async def favicon(request): return web.FileResponse(STATIC_DIR / "favicon.svg") # --------------------------------------------------------------------------- middleware @web.middleware async def security_middleware(request, handler): # Reject cross-site form posts. SameSite=Lax already covers most of it. if request.method in ("POST", "PUT", "DELETE", "PATCH") and request.path.startswith("/api/"): origin = request.headers.get("Origin") if origin: if urlparse(origin).netloc != request.host: return err(403, "Cross-origin request rejected.") response = await handler(request) if request.path == "/" or request.path.startswith("/api/"): response.headers.setdefault("Cache-Control", "no-store") response.headers.setdefault("Pragma", "no-cache") response.headers.setdefault("X-Content-Type-Options", "nosniff") response.headers.setdefault("X-Frame-Options", "DENY") response.headers.setdefault("Referrer-Policy", "no-referrer") response.headers.setdefault( "Content-Security-Policy", "default-src 'self'; " "script-src 'self'; " "style-src 'self' 'unsafe-inline'; " "img-src 'self' data: https://synergia.librus.pl; " "connect-src 'self'; " "font-src 'self'; " "object-src 'none'; " "base-uri 'none'; " "form-action 'self'; " "frame-ancestors 'none'", ) return response @web.middleware async def auth_middleware(request, handler): public = { "/api/config", "/api/session", "/api/login", "/healthz", } session = None token = request.cookies.get(COOKIE_NAME) if token: session = SESSIONS.get(token) if session and session.expired(): SESSIONS.pop(token, None) session = None elif session: session.touch() request["session"] = session if request.path.startswith("/api/") and request.path not in public and session is None: return err(401, "Not authenticated.") return await handler(request) @web.middleware async def error_middleware(request, handler): try: return await handler(request) except ApiError as exc: return err(exc.status, exc.message) except web.HTTPException: raise except json.JSONDecodeError: return err(400, "Malformed JSON body.") except Exception: log.exception("unhandled error on %s %s", request.method, request.path) return err(500, "Internal server error.") # --------------------------------------------------------------------------- app async def healthz(request): return json_response( {"ok": True, "sessions": len(SESSIONS), "version": VERSION} ) async def cleanup_loop(app): try: while True: await asyncio.sleep(600) cleanup_sessions() cutoff = time.time() - 120 for ip in list(LOGIN_ATTEMPTS): LOGIN_ATTEMPTS[ip] = [ t for t in LOGIN_ATTEMPTS[ip] if t > cutoff ] if not LOGIN_ATTEMPTS[ip]: LOGIN_ATTEMPTS.pop(ip, None) except asyncio.CancelledError: pass def create_app(demo=False): app = web.Application(middlewares=[ error_middleware, security_middleware, auth_middleware, ]) app["demo"] = demo app["started"] = time.time() app.router.add_get("/api/config", api_config) app.router.add_get("/api/session", api_session) app.router.add_post("/api/login", api_login) app.router.add_post("/api/logout", api_logout) app.router.add_get("/api/bootstrap", api_bootstrap) app.router.add_get("/api/notifications", api_notifications) app.router.add_get("/api/grades", api_grades) app.router.add_get("/api/timetable", api_timetable) app.router.add_get("/api/exams", api_exams) app.router.add_get("/api/attendances", api_attendances) app.router.add_get("/api/messages", api_messages) app.router.add_get("/api/message", api_message) app.router.add_get("/api/download", api_download) app.router.add_get("/api/freedays", api_freedays) app.router.add_get("/api/conferences", api_conferences) app.router.add_get("/healthz", healthz) app.router.add_get("/", index) app.router.add_get("/favicon.ico", favicon) app.router.add_get("/favicon.svg", favicon) app.router.add_static("/static", STATIC_DIR, show_index=False) async def start_tasks(app): app["cleanup_task"] = asyncio.create_task(cleanup_loop(app)) async def stop_tasks(app): task = app.get("cleanup_task") if task: task.cancel() try: await task except asyncio.CancelledError: pass app.on_startup.append(start_tasks) app.on_cleanup.append(stop_tasks) return app BANNER = r""" _ _ _ _ _____ _ _ ___ | | (_) |__ _ _ _| |_ _ _ |_ _| | | |_ _| | |__ | | '_ \ '_| _| || | | | | |_| || | |____||_|_.__/_| \__|\_,_| |_| \___/|___| """ def main(): parser = argparse.ArgumentParser( description="TUI-styled web frontend for Librus Synergia" ) parser.add_argument("--host", default=os.environ.get("LIBRUS_HOST", "127.0.0.1")) parser.add_argument( "--port", type=int, default=int(os.environ.get("LIBRUS_PORT", "8765")) ) parser.add_argument( "--demo", action="store_true", default=os.environ.get("LIBRUS_DEMO") == "1", help="serve offline sample data instead of contacting Librus", ) parser.add_argument("--debug", action="store_true") args = parser.parse_args() logging.basicConfig( level=logging.DEBUG if args.debug else logging.INFO, format="%(asctime)s %(levelname)-7s %(message)s", ) print(BANNER) if args.demo: print(" DEMO MODE - no Librus account is contacted.\n") print(" http://%s:%d\n" % (args.host, args.port)) print(" Ctrl+C to stop\n") app = create_app(demo=args.demo) web.run_app( app, host=args.host, port=args.port, print=None, access_log=log if args.debug else None, ) if __name__ == "__main__": main()