|
1 | 1 | import os |
2 | | -from typing import Dict |
| 2 | +import json |
| 3 | +from typing import Dict, Callable |
3 | 4 | from dataclasses import is_dataclass, asdict |
4 | 5 | from datetime import date, datetime |
5 | 6 |
|
6 | | -from falcon import App as FalconApp |
7 | 7 |
|
| 8 | +def create_web_app(data: Dict[str, object]) -> Callable: |
| 9 | + static_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "static") |
8 | 10 |
|
9 | | -def create_web_app(data: Dict[str, object]) -> FalconApp: |
10 | | - app = FalconApp() |
| 11 | + def _to_json_serializable(obj): |
| 12 | + """Recursively convert dataclasses and other objects to JSON-serializable dict.""" |
| 13 | + if isinstance(obj, (date, datetime)): |
| 14 | + return obj.isoformat() |
| 15 | + elif is_dataclass(obj) and not isinstance(obj, type): |
| 16 | + return _to_json_serializable(asdict(obj)) |
| 17 | + elif isinstance(obj, dict): |
| 18 | + return {key: _to_json_serializable(value) for key, value in obj.items()} |
| 19 | + elif isinstance(obj, (list, tuple)): |
| 20 | + return [_to_json_serializable(item) for item in obj] |
| 21 | + else: |
| 22 | + return obj |
11 | 23 |
|
12 | | - class StatisticsResource: |
13 | | - def _to_json_serializable(self, obj): |
14 | | - """Recursively convert dataclasses and other objects to JSON-serializable dict.""" |
15 | | - if isinstance(obj, (date, datetime)): |
16 | | - return obj.isoformat() |
17 | | - elif is_dataclass(obj) and not isinstance(obj, type): |
18 | | - return self._to_json_serializable(asdict(obj)) |
19 | | - elif isinstance(obj, dict): |
20 | | - return {key: self._to_json_serializable(value) for key, value in obj.items()} |
21 | | - elif isinstance(obj, (list, tuple)): |
22 | | - return [self._to_json_serializable(item) for item in obj] |
| 24 | + def _serve_static_file(environ, start_response, file_path): |
| 25 | + try: |
| 26 | + with open(file_path, "rb") as f: |
| 27 | + content = f.read() |
| 28 | + |
| 29 | + if file_path.endswith(".html"): |
| 30 | + content_type = "text/html; charset=utf-8" |
| 31 | + elif file_path.endswith(".js"): |
| 32 | + content_type = "application/javascript; charset=utf-8" |
| 33 | + elif file_path.endswith(".css"): |
| 34 | + content_type = "text/css; charset=utf-8" |
| 35 | + elif file_path.endswith(".json"): |
| 36 | + content_type = "application/json; charset=utf-8" |
| 37 | + elif file_path.endswith(".png"): |
| 38 | + content_type = "image/png" |
| 39 | + elif file_path.endswith(".jpg") or file_path.endswith(".jpeg"): |
| 40 | + content_type = "image/jpeg" |
| 41 | + elif file_path.endswith(".svg"): |
| 42 | + content_type = "image/svg+xml" |
| 43 | + elif file_path.endswith(".ico"): |
| 44 | + content_type = "image/x-icon" |
23 | 45 | else: |
24 | | - return obj |
| 46 | + content_type = "application/octet-stream" |
| 47 | + |
| 48 | + start_response("200 OK", [("Content-Type", content_type), ("Content-Length", str(len(content)))]) |
| 49 | + return [content] |
| 50 | + except FileNotFoundError: |
| 51 | + start_response("404 Not Found", [("Content-Type", "text/plain")]) |
| 52 | + return [b"404 - File Not Found"] |
| 53 | + except Exception as e: |
| 54 | + start_response("500 Internal Server Error", [("Content-Type", "text/plain")]) |
| 55 | + return [f"500 - Internal Server Error: {str(e)}".encode("utf-8")] |
| 56 | + |
| 57 | + def wsgi_app(environ, start_response): |
| 58 | + """WSGI application.""" |
| 59 | + path = environ.get("PATH_INFO", "/") |
| 60 | + method = environ.get("REQUEST_METHOD", "GET") |
| 61 | + |
| 62 | + # API endpoint |
| 63 | + if path == "/api/statistics" and method == "GET": |
| 64 | + try: |
| 65 | + serializable_data = _to_json_serializable(data) |
| 66 | + json_data = json.dumps(serializable_data, ensure_ascii=False, indent=2) |
| 67 | + response_body = json_data.encode("utf-8") |
| 68 | + |
| 69 | + start_response( |
| 70 | + "200 OK", |
| 71 | + [("Content-Type", "application/json; charset=utf-8"), ("Content-Length", str(len(response_body)))], |
| 72 | + ) |
| 73 | + return [response_body] |
| 74 | + except Exception as e: |
| 75 | + error_body = json.dumps({"error": str(e)}).encode("utf-8") |
| 76 | + start_response( |
| 77 | + "500 Internal Server Error", |
| 78 | + [("Content-Type", "application/json; charset=utf-8"), ("Content-Length", str(len(error_body)))], |
| 79 | + ) |
| 80 | + return [error_body] |
| 81 | + |
| 82 | + # static files |
| 83 | + if path == "/": |
| 84 | + path = "/index.html" |
| 85 | + |
| 86 | + # Security: prevent path traversal |
| 87 | + path = path.lstrip("/") |
| 88 | + if ".." in path or path.startswith("/"): |
| 89 | + start_response("403 Forbidden", [("Content-Type", "text/plain")]) |
| 90 | + return [b"403 - Forbidden"] |
25 | 91 |
|
26 | | - def on_get(self, req, resp): |
27 | | - resp.media = self._to_json_serializable(data) |
| 92 | + file_path = os.path.join(static_path, path) |
28 | 93 |
|
29 | | - app.add_route("/api/statistics", StatisticsResource()) |
| 94 | + # Security: ensure the file is within the static directory |
| 95 | + if not os.path.abspath(file_path).startswith(os.path.abspath(static_path)): |
| 96 | + start_response("403 Forbidden", [("Content-Type", "text/plain")]) |
| 97 | + return [b"403 - Forbidden"] |
30 | 98 |
|
31 | | - static_path = os.path.dirname(os.path.abspath(__file__)) + "/static/" |
32 | | - app.add_static_route("/", static_path) |
| 99 | + return _serve_static_file(environ, start_response, file_path) |
33 | 100 |
|
34 | | - return app |
| 101 | + return wsgi_app |
0 commit comments