257 lines
10 KiB
Python
257 lines
10 KiB
Python
"""TieMeasureFlow Client - Flask Entry Point."""
|
|
import json
|
|
import os
|
|
import re
|
|
from datetime import date, datetime, timezone
|
|
from zoneinfo import ZoneInfo
|
|
from urllib.parse import urlparse
|
|
|
|
from flask import Flask, flash, g, jsonify, redirect, url_for, session, request
|
|
from flask_babel import Babel, gettext as _
|
|
from flask_wtf.csrf import CSRFError, CSRFProtect
|
|
from markupsafe import Markup, escape
|
|
from werkzeug.middleware.proxy_fix import ProxyFix
|
|
|
|
from config import Config, guard_default_secret_key
|
|
|
|
# **bold**, the whole of the markup a task description understands besides the
|
|
# line break. DOTALL so a phrase that wraps onto the next line still closes.
|
|
_BOLD_RX = re.compile(r"\*\*(.+?)\*\*", re.DOTALL)
|
|
|
|
# Punto 12. Nothing loads from outside: every library ships with the install
|
|
# (static/vendor/), because the shop floor network has no way out and a page that
|
|
# waits for a CDN there is a white screen, not a slow one.
|
|
#
|
|
# unsafe-inline the templates carry their Alpine components inline
|
|
# unsafe-eval Alpine 3 and Plotly both compile expressions at runtime
|
|
# blob: PDF.js runs its worker from a blob, Fabric exports canvases
|
|
# data: thumbnails and canvas exports are inlined
|
|
CSP = (
|
|
"default-src 'self'; "
|
|
"script-src 'self' 'unsafe-inline' 'unsafe-eval'; "
|
|
"style-src 'self' 'unsafe-inline'; "
|
|
"font-src 'self'; "
|
|
"img-src 'self' data: blob:; "
|
|
"connect-src 'self'; "
|
|
"worker-src 'self' blob:; "
|
|
"object-src 'none'; "
|
|
"base-uri 'self'; "
|
|
"form-action 'self'; "
|
|
"frame-ancestors 'none'"
|
|
)
|
|
|
|
|
|
def get_locale():
|
|
"""Get user's preferred language from session or Accept-Language header."""
|
|
# 1. User preference in session
|
|
if "language" in session:
|
|
return session["language"]
|
|
# 2. Browser Accept-Language
|
|
return request.accept_languages.best_match(
|
|
Config.LANGUAGES.keys(), default="it"
|
|
)
|
|
|
|
|
|
def create_app() -> Flask:
|
|
"""Application factory."""
|
|
app = Flask(__name__)
|
|
app.config.from_object(Config)
|
|
|
|
# Trust one reverse-proxy hop (Nginx in dev, Traefik in prod) so that
|
|
# request.remote_addr returns the real tablet IP rather than the proxy IP.
|
|
# The APIClient forwards that IP to FastAPI for accurate rate limiting.
|
|
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1)
|
|
|
|
# Initialize CSRF protection
|
|
csrf = CSRFProtect(app)
|
|
|
|
# Initialize Flask-Babel
|
|
Babel(app, locale_selector=get_locale)
|
|
|
|
# Register blueprints
|
|
from blueprints.auth import auth_bp
|
|
from blueprints.measure import measure_bp
|
|
from blueprints.maker import maker_bp
|
|
from blueprints.statistics import statistics_bp
|
|
from blueprints.admin import admin_bp
|
|
from blueprints.monitor import monitor_bp # #57 live station monitor
|
|
|
|
app.register_blueprint(auth_bp)
|
|
app.register_blueprint(measure_bp, url_prefix="/measure")
|
|
app.register_blueprint(maker_bp, url_prefix="/maker")
|
|
app.register_blueprint(statistics_bp, url_prefix="/statistics")
|
|
app.register_blueprint(admin_bp, url_prefix="/admin")
|
|
app.register_blueprint(monitor_bp, url_prefix="/monitor") # #57
|
|
|
|
@app.before_request
|
|
def _check_secret_key():
|
|
"""See guard_default_secret_key in config.py (#33)."""
|
|
guard_default_secret_key(app)
|
|
|
|
@app.route("/")
|
|
def index():
|
|
"""Root redirect to login or dashboard based on session."""
|
|
if "user" in session:
|
|
return redirect(url_for("measure.select_recipe"))
|
|
return redirect(url_for("auth.login"))
|
|
|
|
@app.route("/set-language/<lang>")
|
|
def set_language(lang):
|
|
"""Set user's preferred language and store in session."""
|
|
if lang in Config.LANGUAGES:
|
|
session["language"] = lang
|
|
# Only follow the referrer if it points back to this host
|
|
# (prevents open redirect via a forged Referer header).
|
|
referrer = request.referrer
|
|
if referrer:
|
|
parsed = urlparse(referrer)
|
|
if parsed.netloc and parsed.netloc != request.host:
|
|
referrer = None
|
|
return redirect(referrer or url_for("auth.login"))
|
|
|
|
@app.errorhandler(CSRFError)
|
|
def handle_csrf_error(error):
|
|
"""CSRF token expired or missing (#12): a stale page, not a network failure.
|
|
|
|
WTF_CSRF_TIME_LIMIT is None so a token only goes bad with the session
|
|
itself, but a tab left open across a forced logout (or missing token)
|
|
still lands here. Fetch/JSON callers get a body they can branch on
|
|
instead of an HTML error page that breaks `response.json()`; page
|
|
loads get a flash and a redirect back to where they were.
|
|
"""
|
|
wants_json = (
|
|
request.accept_mimetypes.best_match(["application/json", "text/html"])
|
|
== "application/json"
|
|
or (request.content_type or "").startswith("application/json")
|
|
or request.headers.get("X-Requested-With") == "XMLHttpRequest"
|
|
)
|
|
if wants_json:
|
|
return (
|
|
jsonify({"error": True, "csrf_expired": True, "detail": error.description}),
|
|
400,
|
|
)
|
|
|
|
flash(_("Sessione scaduta: effettua nuovamente l'accesso."), "error")
|
|
referrer = request.referrer
|
|
if referrer:
|
|
parsed = urlparse(referrer)
|
|
if parsed.netloc and parsed.netloc != request.host:
|
|
referrer = None
|
|
return redirect(referrer or url_for("auth.login"))
|
|
|
|
@app.template_filter("rich_text")
|
|
def rich_text_filter(value):
|
|
"""Render a task description keeping its line breaks and its bold.
|
|
|
|
Whoever writes a recipe pastes from the PDF of the technical sheet, and the
|
|
text used to arrive flattened. Two conventions carry it: a blank line is a
|
|
line break, **like this** is bold.
|
|
|
|
No HTML is ever stored or trusted - the text is escaped first and the only
|
|
tags in the result are the ones produced here. That is the sanitisation:
|
|
there is nothing to sanitise, because nothing is accepted.
|
|
"""
|
|
if not value:
|
|
return Markup("")
|
|
escaped = str(escape(str(value)))
|
|
bolded = _BOLD_RX.sub(r"<strong>\1</strong>", escaped)
|
|
return Markup(bolded.replace("\n", "<br>"))
|
|
|
|
@app.template_filter("local_dt")
|
|
def local_dt_filter(value, fmt="%d/%m/%Y %H:%M"):
|
|
"""Show a UTC datetime (ISO string or datetime) in DISPLAY_TIMEZONE.
|
|
|
|
A value without timezone is taken as UTC. None or empty gives "-".
|
|
"""
|
|
if value is None or value == "":
|
|
return "-"
|
|
if isinstance(value, str):
|
|
try:
|
|
value = datetime.fromisoformat(value.strip().replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return value
|
|
if not isinstance(value, datetime):
|
|
return str(value)
|
|
if value.tzinfo is None:
|
|
value = value.replace(tzinfo=timezone.utc)
|
|
tz = ZoneInfo(app.config.get("DISPLAY_TIMEZONE") or "Europe/Rome")
|
|
return value.astimezone(tz).strftime(fmt)
|
|
|
|
@app.template_filter("tojson_attr")
|
|
def tojson_attr_filter(value):
|
|
"""JSON encode safe for HTML attributes (x-data, etc.).
|
|
|
|
Unlike |tojson, this escapes double quotes to " so the output
|
|
can be safely embedded inside double-quoted HTML attributes.
|
|
The browser decodes the entities before Alpine.js evaluates them.
|
|
"""
|
|
rv = json.dumps(value, ensure_ascii=False)
|
|
rv = (
|
|
rv.replace("&", "\\u0026")
|
|
.replace("<", "\\u003c")
|
|
.replace(">", "\\u003e")
|
|
.replace("'", "\\u0027")
|
|
.replace('"', """)
|
|
)
|
|
return Markup(rv)
|
|
|
|
@app.after_request
|
|
def security_headers(response):
|
|
"""Same-origin only, on the pages the browser actually loads.
|
|
|
|
The policy existed on the backend, which serves the API; the HTML comes
|
|
from here and carried no policy at all. So the rule that says «nothing
|
|
from outside» was written where it could not be broken and absent where
|
|
it could.
|
|
|
|
It is also the guard that makes point 12 stay done: a CDN tag added to a
|
|
template months from now is refused by the browser here, at a desk with a
|
|
network, instead of on a shop floor that has none.
|
|
"""
|
|
response.headers.setdefault("Content-Security-Policy", CSP)
|
|
response.headers.setdefault("X-Content-Type-Options", "nosniff")
|
|
response.headers.setdefault("X-Frame-Options", "DENY")
|
|
response.headers.setdefault("Referrer-Policy", "strict-origin-when-cross-origin")
|
|
return response
|
|
|
|
@app.after_request
|
|
def handle_expired_session(response):
|
|
"""Turn a 401 from the backend into a redirect, for page loads (#25).
|
|
|
|
api_client already emptied the session and set ``g.session_expired``
|
|
on any 401. JSON/AJAX proxy routes already forward the api_client
|
|
dict (which carries ``session_expired``) as their own body, so a
|
|
fetch caller has what it needs. What is left is the page views that
|
|
swallowed the error (flashed something generic and rendered anyway,
|
|
e.g. an empty recipe list): those get replaced here, in the one
|
|
place that sees every response, instead of one check per route.
|
|
"""
|
|
if getattr(g, "session_expired", False) and not response.is_json:
|
|
flash(_("Sessione scaduta: effettua nuovamente l'accesso."), "error")
|
|
return redirect(url_for("auth.login"))
|
|
return response
|
|
|
|
@app.context_processor
|
|
def inject_globals():
|
|
"""Inject global variables into all templates."""
|
|
return {
|
|
"current_user": session.get("user"),
|
|
"current_theme": session.get("theme", "light"),
|
|
"current_language": get_locale(),
|
|
"languages": Config.LANGUAGES,
|
|
"company_logo": session.get("company_logo"),
|
|
"auto_logout_minutes": session.get("auto_logout_minutes"),
|
|
"current_year": date.today().year,
|
|
}
|
|
|
|
return app
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app = create_app()
|
|
app.run(
|
|
host=os.getenv("CLIENT_HOST", "0.0.0.0"),
|
|
port=int(os.getenv("CLIENT_PORT", "5000")),
|
|
debug=True,
|
|
)
|