515 lines
20 KiB
Python
515 lines
20 KiB
Python
import logging
|
|
import secrets
|
|
from html import escape
|
|
from datetime import datetime, timedelta
|
|
from flask import Blueprint, Response, current_app, jsonify, redirect, request, url_for
|
|
from sqlalchemy import select
|
|
|
|
from app import db
|
|
from app.kiosk_qr import pair_url_to_qr_data_uri
|
|
from app.models import KioskDevice, KioskPairToken
|
|
from app.routes.auth_decorators import can_issue_kiosk_access_link
|
|
|
|
bp = Blueprint("kiosk", __name__, url_prefix="/api/kiosk")
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _token_log_fragment(token: str) -> str:
|
|
"""Фрагмент токена для логов (не логируем полностью)."""
|
|
if not token:
|
|
return "(пусто)"
|
|
return (token[:12] + "…") if len(token) > 12 else token
|
|
|
|
|
|
def _device_cookie_id() -> str:
|
|
return (request.cookies.get("wesp_kiosk_device_id") or "").strip()
|
|
|
|
|
|
def _error(message: str, status_code: int = 400):
|
|
return jsonify({"error": True, "message": message}), status_code
|
|
|
|
|
|
def _confirm_html_page(inner_html: str, status: int = 200) -> Response:
|
|
body = (
|
|
"<!DOCTYPE html><html lang=\"ru\"><head><meta charset=\"utf-8\">"
|
|
"<meta name=\"viewport\" content=\"width=device-width, initial-scale=1.0\"></head>"
|
|
f"<body style=\"font-family:system-ui,sans-serif;padding:24px;max-width:520px;margin:0 auto;\">{inner_html}</body></html>"
|
|
)
|
|
return Response(body, status=status, mimetype="text/html; charset=utf-8")
|
|
|
|
|
|
def _confirm_pair_form_html(token: str) -> Response:
|
|
safe = escape(token, quote=True)
|
|
action = escape(url_for("kiosk.confirm_pair"), quote=True)
|
|
inner = (
|
|
"<h1 style=\"font-size:1.25rem;\">Подтвердите привязку терминала</h1>"
|
|
"<p style=\"color:#444;line-height:1.45;\">Нажмите кнопку, чтобы завершить привязку. "
|
|
"Страница без кнопки не расходует токен — так ссылка не «сгорает» из‑за предпросмотра в мессенджере или лишнего запроса.</p>"
|
|
f'<form method="post" action="{action}">'
|
|
f'<input type="hidden" name="token" value="{safe}">'
|
|
'<button type="submit" style="margin-top:12px;padding:14px 22px;font-size:1.05rem;cursor:pointer;'
|
|
'border-radius:8px;border:1px solid #333;background:#1a1a1a;color:#fff;">Подтвердить привязку</button>'
|
|
"</form>"
|
|
)
|
|
return _confirm_html_page(inner, 200)
|
|
|
|
|
|
def _confirm_pair_error_html(message: str, status: int) -> Response:
|
|
inner = f"<p>{escape(message)}</p>"
|
|
return _confirm_html_page(inner, status)
|
|
|
|
|
|
def _scales_public_url() -> str:
|
|
"""Тот же origin, что в QR (`_resolve_public_base_url`), путь /scales."""
|
|
return _resolve_public_base_url().rstrip("/") + "/scales"
|
|
|
|
|
|
def _confirm_pair_success_response():
|
|
"""После привязки — редирект на /scales по LAN/public base; для JSON — поле redirect."""
|
|
target = _scales_public_url()
|
|
if request.is_json or (request.headers.get("Accept") or "").find("application/json") >= 0:
|
|
return jsonify({"success": True, "paired": True, "redirect": target}), 200
|
|
return redirect(target, code=302)
|
|
|
|
|
|
def _confirm_pair_error_response(message: str, status_code: int):
|
|
"""JSON для API, HTML для отправки формы из браузера."""
|
|
if request.is_json or (request.headers.get("Accept") or "").find("application/json") >= 0:
|
|
return _error(message, status_code)
|
|
return _confirm_pair_error_html(message, status_code)
|
|
|
|
|
|
def _is_local_host(host: str) -> bool:
|
|
return host.lower() in {"localhost", "127.0.0.1", "::1"}
|
|
|
|
|
|
from app.services.lan_network import detect_lan_ip
|
|
|
|
|
|
def _resolve_public_base_url() -> str:
|
|
override = str(current_app.config.get("KIOSK_PUBLIC_BASE_URL", "") or "").strip().rstrip("/")
|
|
if override:
|
|
return override
|
|
|
|
host = (request.host or "").strip()
|
|
host_name = host.split(":")[0] if host else ""
|
|
if not _is_local_host(host_name):
|
|
return request.host_url.rstrip("/")
|
|
|
|
lan_ip = detect_lan_ip()
|
|
if not lan_ip:
|
|
return request.host_url.rstrip("/")
|
|
|
|
port = str(request.environ.get("SERVER_PORT") or "").strip()
|
|
# На ферме Flask обычно слушает plain HTTP (80). HTTPS в Start URL без reverse-proxy
|
|
# приводит к «Bad request version» в логах (TLS-handshake на HTTP-порт).
|
|
scheme = "https" if port == "443" else "http"
|
|
if port and port not in {"80", "443"}:
|
|
return f"{scheme}://{lan_ip}:{port}"
|
|
return f"{scheme}://{lan_ip}"
|
|
|
|
|
|
def _ensure_access_slug(device: KioskDevice, *, refresh: bool = False) -> str:
|
|
if device.access_slug and not refresh:
|
|
return device.access_slug
|
|
device.access_slug = secrets.token_urlsafe(32)
|
|
db.session.commit()
|
|
return device.access_slug
|
|
|
|
|
|
def _start_url_for_slug(slug: str) -> str:
|
|
return f"{_resolve_public_base_url().rstrip('/')}/kiosk/start/{slug}"
|
|
|
|
|
|
def _access_link_refresh_requested() -> bool:
|
|
if request.args.get("refresh") == "1":
|
|
return True
|
|
payload = request.get_json(silent=True) or {}
|
|
return bool(payload.get("refresh"))
|
|
|
|
|
|
def _activate_by_token(token: str):
|
|
row = db.session.execute(select(KioskPairToken).where(KioskPairToken.token == token)).scalar_one_or_none()
|
|
if not row:
|
|
logger.warning(
|
|
"[KIOSK] токен не найден token=%s remote=%s",
|
|
_token_log_fragment(token),
|
|
request.remote_addr,
|
|
)
|
|
return None, _error("Токен не найден", 404)
|
|
if row.used_at is not None:
|
|
logger.warning(
|
|
"[KIOSK] токен уже использован token=%s device_id_в_токене=%s remote=%s",
|
|
_token_log_fragment(token),
|
|
row.device_id,
|
|
request.remote_addr,
|
|
)
|
|
return None, _error("Токен уже использован", 409)
|
|
if row.expires_at < datetime.utcnow():
|
|
logger.warning(
|
|
"[KIOSK] токен истёк token=%s remote=%s",
|
|
_token_log_fragment(token),
|
|
request.remote_addr,
|
|
)
|
|
return None, _error("Срок действия токена истёк", 410)
|
|
return row, None
|
|
|
|
|
|
def _pair_current_device_by_token(token: str):
|
|
target_device_id = _device_cookie_id()
|
|
if not target_device_id:
|
|
logger.warning(
|
|
"[KIOSK] claim/scan: нет cookie wesp_kiosk_device_id token=%s remote=%s host=%s",
|
|
_token_log_fragment(token),
|
|
request.remote_addr,
|
|
request.host,
|
|
)
|
|
return None, _error("Не найден cookie терминала. Откройте kiosk-страницу заново.", 400)
|
|
target_device = db.session.get(KioskDevice, target_device_id)
|
|
if not target_device:
|
|
logger.warning(
|
|
"[KIOSK] claim/scan: нет строки KioskDevice для cookie device_id=%s",
|
|
target_device_id,
|
|
)
|
|
return None, _error("Терминал не зарегистрирован на сервере.", 404)
|
|
|
|
row, error = _activate_by_token(token)
|
|
if error:
|
|
return None, error
|
|
|
|
row.used_at = datetime.utcnow()
|
|
target_device.status = "active"
|
|
target_device.last_seen_at = datetime.utcnow()
|
|
target_device.last_ip = request.remote_addr
|
|
db.session.commit()
|
|
logger.info(
|
|
"[KIOSK] claim/scan OK: активирован device_id=%s token=%s remote=%s",
|
|
target_device.id,
|
|
_token_log_fragment(token),
|
|
request.remote_addr,
|
|
)
|
|
return target_device, None
|
|
|
|
|
|
@bp.get("/status")
|
|
def kiosk_status():
|
|
enforce_paired_only = bool(current_app.config.get("KIOSK_ENFORCE_PAIRED_ONLY", True))
|
|
device_id = _device_cookie_id()
|
|
if not device_id:
|
|
logger.debug("[KIOSK] /status без cookie remote=%s host=%s", request.remote_addr, request.host)
|
|
return jsonify(
|
|
{
|
|
"paired": False,
|
|
"device_id": None,
|
|
"status": "unregistered",
|
|
"enforce_paired_only": enforce_paired_only,
|
|
}
|
|
)
|
|
device = db.session.get(KioskDevice, device_id)
|
|
if not device:
|
|
logger.debug(
|
|
"[KIOSK] /status неизвестный device_id=%s remote=%s",
|
|
device_id,
|
|
request.remote_addr,
|
|
)
|
|
return jsonify(
|
|
{
|
|
"paired": False,
|
|
"device_id": device_id,
|
|
"status": "unregistered",
|
|
"enforce_paired_only": enforce_paired_only,
|
|
}
|
|
)
|
|
paired = device.status == "active"
|
|
if paired:
|
|
device.last_seen_at = datetime.utcnow()
|
|
device.last_ip = request.remote_addr
|
|
db.session.commit()
|
|
logger.debug(
|
|
"[KIOSK] /status device_id=%s paired=%s status=%s remote=%s",
|
|
device_id,
|
|
paired,
|
|
device.status,
|
|
request.remote_addr,
|
|
)
|
|
return jsonify(
|
|
{
|
|
"paired": paired,
|
|
"device_id": device.id,
|
|
"status": device.status,
|
|
"display_name": device.display_name,
|
|
"enforce_paired_only": enforce_paired_only,
|
|
}
|
|
)
|
|
|
|
|
|
@bp.get("/setup-available")
|
|
def kiosk_setup_available():
|
|
"""Можно ли показывать UI выдачи Start URL (localhost или admin-сессия)."""
|
|
return jsonify({"can_issue_access_link": can_issue_kiosk_access_link()})
|
|
|
|
|
|
@bp.route("/access-link", methods=["GET", "POST"])
|
|
def kiosk_access_link():
|
|
"""Постоянная Start URL для Fully Kiosk + QR."""
|
|
if not can_issue_kiosk_access_link():
|
|
logger.warning(
|
|
"[KIOSK] access-link: отказ (не localhost и нет admin-сессии) remote=%s host=%s",
|
|
request.remote_addr,
|
|
request.host,
|
|
)
|
|
return _error("Выдача Start URL доступна только администратору или с localhost.", 403)
|
|
|
|
device_id = _device_cookie_id()
|
|
if not device_id:
|
|
logger.warning(
|
|
"[KIOSK] access-link: нет cookie remote=%s host=%s",
|
|
request.remote_addr,
|
|
request.host,
|
|
)
|
|
return _error("Не найден cookie терминала. Откройте kiosk-страницу заново.", 400)
|
|
device = db.session.get(KioskDevice, device_id)
|
|
if not device:
|
|
logger.warning("[KIOSK] access-link: нет KioskDevice для id=%s", device_id)
|
|
return _error("Терминал не зарегистрирован на сервере.", 404)
|
|
|
|
refresh = _access_link_refresh_requested()
|
|
slug = _ensure_access_slug(device, refresh=refresh)
|
|
start_url = _start_url_for_slug(slug)
|
|
try:
|
|
qr_img_url = pair_url_to_qr_data_uri(start_url)
|
|
except Exception:
|
|
logger.exception(
|
|
"[KIOSK] access-link: ошибка генерации QR device_id=%s",
|
|
device.id,
|
|
)
|
|
return _error("Не удалось сгенерировать QR-код", 500)
|
|
|
|
logger.info(
|
|
"[KIOSK] access-link: device_id=%s refresh=%s public_base=%s remote=%s",
|
|
device.id,
|
|
refresh,
|
|
_resolve_public_base_url(),
|
|
request.remote_addr,
|
|
)
|
|
return jsonify(
|
|
{
|
|
"success": True,
|
|
"start_url": start_url,
|
|
"pair_url": start_url,
|
|
"qr_image_url": qr_img_url,
|
|
"device_id": device.id,
|
|
"permanent": True,
|
|
}
|
|
)
|
|
|
|
|
|
@bp.post("/pair-token")
|
|
def kiosk_pair_token():
|
|
if not can_issue_kiosk_access_link():
|
|
return _error("Выдача pair-token доступна только администратору или с localhost.", 403)
|
|
|
|
device_id = _device_cookie_id()
|
|
if not device_id:
|
|
logger.warning(
|
|
"[KIOSK] pair-token: нет cookie remote=%s host=%s",
|
|
request.remote_addr,
|
|
request.host,
|
|
)
|
|
return _error("Не найден cookie терминала. Откройте kiosk-страницу заново.", 400)
|
|
device = db.session.get(KioskDevice, device_id)
|
|
if not device:
|
|
logger.warning("[KIOSK] pair-token: нет KioskDevice для id=%s", device_id)
|
|
return _error("Терминал не зарегистрирован на сервере.", 404)
|
|
|
|
ttl = max(30, int(current_app.config.get("KIOSK_PAIR_TOKEN_TTL_SECONDS", 300)))
|
|
token = secrets.token_urlsafe(24)
|
|
obj = KioskPairToken(
|
|
token=token,
|
|
device_id=device.id,
|
|
expires_at=datetime.utcnow() + timedelta(seconds=ttl),
|
|
issued_for="kiosk_qr_pair",
|
|
)
|
|
db.session.add(obj)
|
|
db.session.commit()
|
|
|
|
base = _resolve_public_base_url()
|
|
# QR и pair_url ведут на confirm: привязка киоска по device_id из токена, без cookie на телефоне.
|
|
# pair/claim остаётся для сценария «второй терминал уже открыл сайт и имеет свой cookie».
|
|
confirm_url = f"{base}/api/kiosk/pair/confirm?token={token}"
|
|
pair_url = confirm_url
|
|
try:
|
|
qr_img_url = pair_url_to_qr_data_uri(pair_url)
|
|
except Exception:
|
|
logger.exception(
|
|
"[KIOSK] pair-token: ошибка локальной генерации QR device_id=%s",
|
|
device.id,
|
|
)
|
|
return _error("Не удалось сгенерировать QR-код", 500)
|
|
logger.info(
|
|
"[KIOSK] pair-token выдан: device_id=%s token=%s public_base=%s remote=%s host=%s",
|
|
device.id,
|
|
_token_log_fragment(token),
|
|
base,
|
|
request.remote_addr,
|
|
request.host,
|
|
)
|
|
return jsonify(
|
|
{
|
|
"success": True,
|
|
"token": token,
|
|
"pair_url": pair_url,
|
|
"confirm_url": confirm_url,
|
|
"qr_image_url": qr_img_url,
|
|
"expires_in": ttl,
|
|
}
|
|
)
|
|
|
|
|
|
@bp.route("/pair/confirm", methods=["GET", "POST"])
|
|
def confirm_pair():
|
|
"""GET — страница с кнопкой (токен не расходуется). POST — фактическая привязка."""
|
|
if request.method == "GET":
|
|
token = (request.args.get("token") or "").strip()
|
|
if not token:
|
|
logger.warning("[KIOSK] pair/confirm GET без token remote=%s", request.remote_addr)
|
|
return _confirm_pair_error_html("token обязателен", 400)
|
|
|
|
row = db.session.execute(
|
|
select(KioskPairToken).where(KioskPairToken.token == token)
|
|
).scalar_one_or_none()
|
|
if not row:
|
|
logger.warning(
|
|
"[KIOSK] pair/confirm GET токен не найден token=%s remote=%s",
|
|
_token_log_fragment(token),
|
|
request.remote_addr,
|
|
)
|
|
return _confirm_pair_error_html("Токен не найден", 404)
|
|
if row.used_at is not None:
|
|
inner = (
|
|
"<h2>Привязка уже выполнена</h2>"
|
|
"<p>Этот одноразовый код уже использован. Если киоск не показывает доступ, "
|
|
"обновите страницу весов на терминале.</p>"
|
|
)
|
|
return _confirm_html_page(inner, 200)
|
|
if row.expires_at < datetime.utcnow():
|
|
logger.warning(
|
|
"[KIOSK] pair/confirm GET токен просрочен token=%s remote=%s",
|
|
_token_log_fragment(token),
|
|
request.remote_addr,
|
|
)
|
|
return _confirm_pair_error_html("Срок действия токена истёк", 410)
|
|
logger.info(
|
|
"[KIOSK] pair/confirm GET форма подтверждения token=%s device_id_в_токене=%s remote=%s",
|
|
_token_log_fragment(token),
|
|
row.device_id,
|
|
request.remote_addr,
|
|
)
|
|
return _confirm_pair_form_html(token)
|
|
|
|
token = (request.form.get("token") or "").strip()
|
|
if not token:
|
|
payload = request.get_json(silent=True) or {}
|
|
token = str(payload.get("token") or "").strip()
|
|
if not token:
|
|
return _confirm_pair_error_response("token обязателен", 400)
|
|
|
|
row, error = _activate_by_token(token)
|
|
if error:
|
|
resp, status_code = error
|
|
data = resp.get_json(silent=True) or {}
|
|
msg = data.get("message", "Ошибка")
|
|
logger.warning(
|
|
"[KIOSK] pair/confirm POST отказ: %s token=%s remote=%s status=%s",
|
|
msg,
|
|
_token_log_fragment(token),
|
|
request.remote_addr,
|
|
status_code,
|
|
)
|
|
return _confirm_pair_error_response(msg, status_code)
|
|
|
|
device = db.session.get(KioskDevice, row.device_id)
|
|
if not device:
|
|
logger.error(
|
|
"[KIOSK] pair/confirm POST: нет KioskDevice для device_id=%s из токена",
|
|
row.device_id,
|
|
)
|
|
return _confirm_pair_error_response("Терминал не найден", 404)
|
|
|
|
row.used_at = datetime.utcnow()
|
|
device.status = "active"
|
|
device.last_seen_at = datetime.utcnow()
|
|
device.last_ip = request.remote_addr
|
|
|
|
# Телефон с другим хостом, чем главный киоск (localhost vs LAN), имеет свой cookie и
|
|
# отдельный KioskDevice. Редирект на /scales иначе попадает в guard с pending — активируем
|
|
# и браузер, который отправил POST, если это другой pending-терминал.
|
|
helper_id = _device_cookie_id()
|
|
if helper_id and helper_id != device.id:
|
|
helper = db.session.get(KioskDevice, helper_id)
|
|
if helper and helper.status == "pending":
|
|
helper.status = "active"
|
|
helper.last_seen_at = datetime.utcnow()
|
|
helper.last_ip = request.remote_addr
|
|
logger.info(
|
|
"[KIOSK] pair/confirm POST: также активирован браузер подтверждения device_id=%s "
|
|
"(терминал из токена=%s)",
|
|
helper_id,
|
|
device.id,
|
|
)
|
|
|
|
db.session.commit()
|
|
redirect_to = _scales_public_url()
|
|
logger.info(
|
|
"[KIOSK] pair/confirm POST OK: активирован device_id=%s token=%s redirect=%s remote=%s",
|
|
device.id,
|
|
_token_log_fragment(token),
|
|
redirect_to,
|
|
request.remote_addr,
|
|
)
|
|
return _confirm_pair_success_response()
|
|
|
|
|
|
@bp.post("/pair/scan-confirm")
|
|
def confirm_pair_from_scan():
|
|
payload = request.get_json(silent=True) or {}
|
|
token = str(payload.get("token") or "").strip()
|
|
if not token:
|
|
return _error("token обязателен", 400)
|
|
|
|
target_device, error = _pair_current_device_by_token(token)
|
|
if error:
|
|
return error
|
|
|
|
logger.info(
|
|
"[KIOSK] pair/scan-confirm OK device_id=%s token=%s",
|
|
target_device.id,
|
|
_token_log_fragment(token),
|
|
)
|
|
return jsonify({"success": True, "paired": True, "device_id": target_device.id})
|
|
|
|
|
|
@bp.get("/pair/claim")
|
|
def claim_pair_from_link():
|
|
token = (request.args.get("token") or "").strip()
|
|
if not token:
|
|
return _error("token обязателен", 400)
|
|
|
|
logger.info(
|
|
"[KIOSK] pair/claim GET token=%s remote=%s",
|
|
_token_log_fragment(token),
|
|
request.remote_addr,
|
|
)
|
|
target_device, error = _pair_current_device_by_token(token)
|
|
if error:
|
|
return error
|
|
|
|
return (
|
|
"<html><head><meta charset='utf-8'></head><body>"
|
|
"<h2>Терминал привязан</h2>"
|
|
"<p>Переход на экран весов через 3 секунды...</p>"
|
|
"<script>setTimeout(function(){ window.location.href = '/scales'; }, 3000);</script>"
|
|
"</body></html>",
|
|
200,
|
|
)
|