Нажмите кнопку, чтобы завершить привязку. " "Страница без кнопки не расходует токен — так ссылка не «сгорает» из‑за предпросмотра в мессенджере или лишнего запроса.
" f'" ) return _confirm_html_page(inner, 200) def _confirm_pair_error_html(message: str, status: int) -> Response: inner = f"{escape(message)}
" 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 = ( "Этот одноразовый код уже использован. Если киоск не показывает доступ, " "обновите страницу весов на терминале.
" ) 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 ( "" "Переход на экран весов через 3 секунды...
" "" "", 200, )