From 0590247c6bac294824d75fea7b4bf5c99c239e35 Mon Sep 17 00:00:00 2001 From: ProfiAnton Date: Tue, 31 Mar 2026 20:21:27 +0200 Subject: [PATCH] use root path everywhere --- func_to_web/auth.py | 46 +++++--- func_to_web/routes.py | 194 ++++++++++++++++++------------- func_to_web/templates/form.html | 16 +-- func_to_web/templates/index.html | 14 +-- func_to_web/templates/login.html | 92 +++++++++------ 5 files changed, 208 insertions(+), 154 deletions(-) diff --git a/func_to_web/auth.py b/func_to_web/auth.py index f18291d..e72ea1b 100644 --- a/func_to_web/auth.py +++ b/func_to_web/auth.py @@ -5,9 +5,11 @@ from starlette.middleware.sessions import SessionMiddleware -def setup_auth_middleware(app, auth: dict[str, str], templates: Jinja2Templates, secret_key: str | None = None): +def setup_auth_middleware( + app, auth: dict[str, str], templates: Jinja2Templates, secret_key: str | None = None +): """Setup authentication middleware and routes. - + Args: app: FastAPI application instance. auth: Dictionary of {username: password} for authentication. @@ -15,24 +17,34 @@ def setup_auth_middleware(app, auth: dict[str, str], templates: Jinja2Templates, secret_key: Secret key for session signing. """ key = secret_key or secrets.token_hex(32) - + + def _root_path(request: Request): + root_path = request.scope.get("root_path", "") or "" + if root_path.endswith("/"): + root_path = root_path[:-1] + return root_path + + def _with_root_path(request: Request, path: str) -> str: + if not path.startswith("/"): + path = f"/{path}" + return f"{_root_path(request)}{path}" + # 1. Define Auth Middleware (INNER) @app.middleware("http") async def auth_middleware(request: Request, call_next): - path = request.url.path - + path = request.url.path.removeprefix(_root_path(request)) # Allow public paths: login page, auth endpoint, and static assets if path in ["/login", "/auth"] or path.startswith("/static"): return await call_next(request) - + # Check for valid session if not request.session.get("user"): # If API call (AJAX), return 401 if "application/json" in request.headers.get("accept", ""): return JSONResponse({"error": "Unauthorized"}, status_code=401) # If browser navigation, redirect to login - return RedirectResponse(url="/login") - + return RedirectResponse(url=_with_root_path(request, "/login")) + return await call_next(request) # 2. Add SessionMiddleware (OUTER - runs first) @@ -42,7 +54,7 @@ async def auth_middleware(request: Request, call_next): async def login_page(request: Request): # If already logged in, go home if request.session.get("user"): - return RedirectResponse(url="/") + return RedirectResponse(url=_with_root_path(request, "/")) return templates.TemplateResponse("login.html", {"request": request}) @app.post("/auth") @@ -51,24 +63,24 @@ async def authenticate(request: Request): form = await request.form() username = form.get("username") password = form.get("password") - + if username in auth: # Safe comparison against Timing Attacks if secrets.compare_digest(auth[username], password): request.session["user"] = username - return RedirectResponse(url="/", status_code=303) - + return RedirectResponse( + url=_with_root_path(request, "/"), status_code=303 + ) + return templates.TemplateResponse( - "login.html", - {"request": request, "error": "Invalid credentials"} + "login.html", {"request": request, "error": "Invalid credentials"} ) except Exception: return templates.TemplateResponse( - "login.html", - {"request": request, "error": "Login failed"} + "login.html", {"request": request, "error": "Login failed"} ) @app.get("/logout") async def logout(request: Request): request.session.clear() - return RedirectResponse(url="/login") \ No newline at end of file + return RedirectResponse(url=_with_root_path(request, "/login")) diff --git a/func_to_web/routes.py b/func_to_web/routes.py index ca375fc..0f5bca6 100644 --- a/func_to_web/routes.py +++ b/func_to_web/routes.py @@ -4,7 +4,7 @@ from typing import Callable import asyncio -from fastapi import Request +from fastapi import FastAPI, Request from fastapi.responses import JSONResponse, FileResponse as FastAPIFileResponse from fastapi.templating import Jinja2Templates @@ -18,58 +18,60 @@ cleanup_uploaded_file, get_returned_file, cleanup_returned_file, - create_response_with_files + create_response_with_files, ) -UUID_PATTERN = re.compile(r'^[a-f0-9]{32}$') +UUID_PATTERN = re.compile(r"^[a-f0-9]{32}$") async def handle_form_submission( - request: Request, - func: Callable, - params: dict[str, ParamInfo] + request: Request, func: Callable, params: dict[str, ParamInfo] ) -> JSONResponse: """Handle form submission for any function. - + Args: request: FastAPI request object. func: Function to call with validated parameters. params: Parameter metadata from analyze(). - + Returns: JSON response with result or error. """ uploaded_files = [] - + try: form_data = await request.form() data = {} - + for name, info in params.items(): if info.is_list: raw_values = form_data.getlist(name) - + if not raw_values and name in form_data: - raw_values = [form_data[name]] + raw_values = [form_data[name]] processed_list = [] for val in raw_values: - if hasattr(val, 'filename'): + if hasattr(val, "filename"): suffix = os.path.splitext(val.filename)[1] file_path = await save_uploaded_file(val, suffix) uploaded_files.append(file_path) processed_list.append(file_path) else: processed_list.append(val) - - if len(processed_list) == 1 and isinstance(processed_list[0], str) and processed_list[0].startswith('['): + + if ( + len(processed_list) == 1 + and isinstance(processed_list[0], str) + and processed_list[0].startswith("[") + ): data[name] = processed_list[0] else: data[name] = processed_list else: value = form_data.get(name) - if hasattr(value, 'filename'): + if hasattr(value, "filename"): suffix = os.path.splitext(value.filename)[1] file_path = await save_uploaded_file(value, suffix) uploaded_files.append(file_path) @@ -78,78 +80,80 @@ async def handle_form_submission( data[name] = value for key, value in form_data.items(): - if key.endswith('_optional_toggle'): + if key.endswith("_optional_toggle"): data[key] = value - + validated = validate_params(data, params) - + if inspect.iscoroutinefunction(func): result = await func(**validated) else: result = await asyncio.to_thread(func, **validated) - + if config.AUTO_DELETE_UPLOADS: for file_path in uploaded_files: cleanup_uploaded_file(file_path) - + processed = await asyncio.to_thread(process_result, result) response = create_response_with_files(processed) - + return JSONResponse(response) - + except Exception as e: if config.AUTO_DELETE_UPLOADS: for file_path in uploaded_files: cleanup_uploaded_file(file_path) - + return JSONResponse({"success": False, "error": str(e)}, status_code=400) -def setup_download_route(app): +def setup_download_route(app: FastAPI): """Setup file download route. - + Args: app: FastAPI application instance. """ + @app.get("/download/{file_id}") async def download_file(file_id: str): if not UUID_PATTERN.match(file_id): return JSONResponse({"error": "Invalid file ID"}, status_code=400) - + file_info = get_returned_file(file_id) - + if not file_info: return JSONResponse({"error": "File not found"}, status_code=404) - - path = file_info['path'] - filename = file_info['filename'] - + + path = file_info["path"] + filename = file_info["filename"] + if not os.path.exists(path): cleanup_returned_file(file_id, delete_from_disk=False) return JSONResponse({"error": "File expired"}, status_code=404) - + safe_filename = os.path.basename(filename) - + response = FastAPIFileResponse( - path=path, - filename=safe_filename, - media_type='application/octet-stream' + path=path, filename=safe_filename, media_type="application/octet-stream" ) - + return response -def _register_function_routes(app, func: Callable, templates: Jinja2Templates, has_auth: bool): +def _register_function_routes( + app: FastAPI, func: Callable, templates: Jinja2Templates, has_auth: bool +): """Register GET and POST routes for a function.""" params = analyze(func) - func_name = func.__name__.replace('_', ' ').title() + func_name = func.__name__.replace("_", " ").title() description = inspect.getdoc(func) route = f"/{func.__name__}" submit_route = f"{route}/submit" - - def make_form_handler(title: str, prms: dict, desc: str | None, submit_path: str): + + def make_form_handler(title: str, prms: dict, desc: str | None, submit_name: str): async def form_view(request: Request): flds = build_form_fields(prms) + submit_url = request.url_for(submit_name) return templates.TemplateResponse( "form.html", { @@ -157,25 +161,37 @@ async def form_view(request: Request): "title": title, "description": desc, "fields": flds, - "submit_url": submit_path, + "submit_url": submit_url, "show_back_button": True, - "has_auth": has_auth - } + "has_auth": has_auth, + }, ) + return form_view - + def make_submit_handler(fn: Callable, prms: dict): async def submit_view(request: Request): return await handle_form_submission(request, fn, prms) + return submit_view - - app.get(route)(make_form_handler(func_name, params, description, submit_route)) - app.post(submit_route)(make_submit_handler(func, params)) + form_name = f"form_{func.__name__}" + submit_name = f"submit_{func.__name__}" + app.get(route, name=form_name)( + make_form_handler(func_name, params, description, submit_name) + ) + app.post(submit_route, name=submit_name)(make_submit_handler(func, params)) -def setup_single_function_routes(app, func: Callable, params: dict, templates: Jinja2Templates, has_auth: bool): + +def setup_single_function_routes( + app: FastAPI, + func: Callable, + params: dict, + templates: Jinja2Templates, + has_auth: bool, +): """Setup routes for single function mode. - + Args: app: FastAPI application instance. func: The function to wrap. @@ -183,10 +199,10 @@ def setup_single_function_routes(app, func: Callable, params: dict, templates: J templates: Jinja2Templates instance. has_auth: Whether authentication is enabled. """ - func_name = func.__name__.replace('_', ' ').title() + func_name = func.__name__.replace("_", " ").title() description = inspect.getdoc(func) - - @app.get("/") + + @app.get("/", name="index") async def form(request: Request): fields = build_form_fields(params) return templates.TemplateResponse( @@ -196,69 +212,79 @@ async def form(request: Request): "title": func_name, "description": description, "fields": fields, - "submit_url": "/submit", + "submit_url": request.url_for("submit_single"), "show_back_button": False, - "has_auth": has_auth - } + "has_auth": has_auth, + }, ) - @app.post("/submit") + @app.post("/submit", name="submit_single") async def submit(request: Request): return await handle_form_submission(request, func, params) -def setup_multiple_function_routes(app, funcs: list[Callable], templates: Jinja2Templates, has_auth: bool): +def setup_multiple_function_routes( + app: FastAPI, funcs: list[Callable], templates: Jinja2Templates, has_auth: bool +): """Setup routes for multiple functions mode. - + Args: app: FastAPI application instance. funcs: List of functions to wrap. templates: Jinja2Templates instance. has_auth: Whether authentication is enabled. """ - @app.get("/") + + @app.get("/", name="index") async def index(request: Request): - tools = [{ - "name": f.__name__.replace('_', ' ').title(), - "path": f"/{f.__name__}" - } for f in funcs] + tools = [ + { + "name": f.__name__.replace("_", " ").title(), + "path": request.url_for(f"form_{f.__name__}"), + } + for f in funcs + ] return templates.TemplateResponse( - "index.html", - {"request": request, "tools": tools, "has_auth": has_auth} + "index.html", {"request": request, "tools": tools, "has_auth": has_auth} ) - + for func in funcs: _register_function_routes(app, func, templates, has_auth) -def setup_grouped_function_routes(app, grouped_funcs: dict[str, list[Callable]], templates: Jinja2Templates, has_auth: bool): +def setup_grouped_function_routes( + app: FastAPI, + grouped_funcs: dict[str, list[Callable]], + templates: Jinja2Templates, + has_auth: bool, +): """Setup routes for grouped functions mode. - + Args: app: FastAPI application instance. grouped_funcs: Dictionary of {group_name: [functions]}. templates: Jinja2Templates instance. has_auth: Whether authentication is enabled. """ - @app.get("/") + + @app.get("/", name="index") async def index(request: Request): groups = [] for group_name, funcs in grouped_funcs.items(): - tools = [{ - "name": f.__name__.replace('_', ' ').title(), - "path": f"/{f.__name__}" - } for f in funcs] - - groups.append({ - "name": group_name, - "tools": tools - }) - + tools = [ + { + "name": f.__name__.replace("_", " ").title(), + "path": request.url_for(f"form_{f.__name__}"), + } + for f in funcs + ] + + groups.append({"name": group_name, "tools": tools}) + return templates.TemplateResponse( - "index.html", - {"request": request, "groups": groups, "has_auth": has_auth} + "index.html", {"request": request, "groups": groups, "has_auth": has_auth} ) for funcs in grouped_funcs.values(): for func in funcs: - _register_function_routes(app, func, templates, has_auth) \ No newline at end of file + _register_function_routes(app, func, templates, has_auth) diff --git a/func_to_web/templates/form.html b/func_to_web/templates/form.html index 2251227..701abe5 100644 --- a/func_to_web/templates/form.html +++ b/func_to_web/templates/form.html @@ -4,7 +4,7 @@ {{ title }} - + @@ -21,7 +21,7 @@ {% if has_auth %} - + @@ -32,7 +32,7 @@ {% if show_back_button %} - + @@ -277,12 +277,12 @@

{{ title }}

- - - - + + + + - \ No newline at end of file + diff --git a/func_to_web/templates/index.html b/func_to_web/templates/index.html index a439166..85b614c 100644 --- a/func_to_web/templates/index.html +++ b/func_to_web/templates/index.html @@ -4,7 +4,7 @@ Menu - + @@ -21,7 +21,7 @@ {% if has_auth %} - + @@ -74,10 +74,10 @@

{{ group.name }}

{% endif %} - - - - + + + + - \ No newline at end of file + diff --git a/func_to_web/templates/login.html b/func_to_web/templates/login.html index 7713223..575f8ae 100644 --- a/func_to_web/templates/login.html +++ b/func_to_web/templates/login.html @@ -1,42 +1,58 @@ - + - - - - Login - - - - - - - \ No newline at end of file + + +