467 lines
18 KiB
Python
467 lines
18 KiB
Python
import base64
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from app import create_app, db
|
|
from config import TestingConfig
|
|
|
|
|
|
class KioskPairConfig(TestingConfig):
|
|
_TMP_DIR = tempfile.mkdtemp(prefix="wesp-kiosk-pair-tests-")
|
|
SQLALCHEMY_DATABASE_URI = f"sqlite:///{os.path.join(_TMP_DIR, 'recipes_test.db')}"
|
|
SQLALCHEMY_BINDS = {"reports": f"sqlite:///{os.path.join(_TMP_DIR, 'reports_test.db')}"}
|
|
SIMULATION_MODE = True
|
|
KIOSK_ENFORCE_PAIRED_ONLY = True
|
|
KIOSK_PAIR_TOKEN_TTL_SECONDS = 120
|
|
KIOSK_PUBLIC_BASE_URL = ""
|
|
|
|
|
|
class KioskPairingTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.app = create_app(KioskPairConfig)
|
|
self.client = self.app.test_client()
|
|
self.ctx = self.app.app_context()
|
|
self.ctx.push()
|
|
db.create_all()
|
|
self.client.get("/scales")
|
|
|
|
def tearDown(self) -> None:
|
|
from app.routes import scales as scales_module
|
|
|
|
reader = getattr(scales_module, "_scales_reader", None)
|
|
if reader is not None:
|
|
reader.stop()
|
|
scales_module._scales_reader = None
|
|
db.session.remove()
|
|
db.drop_all()
|
|
self.ctx.pop()
|
|
|
|
def _confirm_post(self, token: str):
|
|
return self.client.post(
|
|
"/api/kiosk/pair/confirm",
|
|
data={"token": token},
|
|
content_type="application/x-www-form-urlencoded",
|
|
)
|
|
|
|
def _pair(self) -> str:
|
|
pair = self.client.post("/api/kiosk/pair-token")
|
|
self.assertEqual(pair.status_code, 200)
|
|
token = pair.get_json().get("token")
|
|
self.assertTrue(token)
|
|
page = self.client.get(f"/api/kiosk/pair/confirm?token={token}")
|
|
self.assertEqual(page.status_code, 200)
|
|
self.assertIn("Подтвердить привязку", page.get_data(as_text=True))
|
|
ok = self._confirm_post(token)
|
|
self.assertEqual(ok.status_code, 302, msg=ok.location)
|
|
return token
|
|
|
|
def _non_local_get(self, client, path: str):
|
|
return client.get(
|
|
path,
|
|
base_url="http://example.com",
|
|
environ_overrides={"REMOTE_ADDR": "10.0.0.55"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
def _access_link_payload(self):
|
|
resp = self.client.get("/api/kiosk/access-link")
|
|
self.assertEqual(resp.status_code, 200)
|
|
return resp.get_json()
|
|
|
|
def _start_path_from_access_link(self) -> str:
|
|
payload = self._access_link_payload()
|
|
start_url = payload.get("start_url") or ""
|
|
self.assertTrue(start_url.startswith("http"))
|
|
return start_url.split("/kiosk/start/", 1)[1]
|
|
|
|
def _pair_via_start_url(self, client=None):
|
|
client = client or self.client
|
|
slug = self._start_path_from_access_link()
|
|
return client.get(
|
|
f"/kiosk/start/{slug}",
|
|
base_url="http://example.com",
|
|
environ_overrides={"REMOTE_ADDR": "10.0.0.55"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
def test_access_link_denied_for_external_without_auth(self) -> None:
|
|
ext = self.app.test_client()
|
|
ext.get(
|
|
"/scales",
|
|
base_url="http://example.com",
|
|
environ_overrides={"REMOTE_ADDR": "10.0.0.55"},
|
|
follow_redirects=False,
|
|
)
|
|
resp = ext.get(
|
|
"/api/kiosk/access-link",
|
|
base_url="http://example.com",
|
|
environ_overrides={"REMOTE_ADDR": "10.0.0.55"},
|
|
)
|
|
self.assertEqual(resp.status_code, 403)
|
|
|
|
def test_setup_available_false_for_external(self) -> None:
|
|
resp = self.client.get(
|
|
"/api/kiosk/setup-available",
|
|
base_url="http://example.com",
|
|
environ_overrides={"REMOTE_ADDR": "10.0.0.55"},
|
|
)
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertFalse(resp.get_json().get("can_issue_access_link"))
|
|
|
|
def test_setup_available_true_for_localhost(self) -> None:
|
|
resp = self.client.get("/api/kiosk/setup-available", base_url="http://localhost:5000")
|
|
self.assertTrue(resp.get_json().get("can_issue_access_link"))
|
|
|
|
def test_setup_available_true_with_admin_session(self) -> None:
|
|
with self.client.session_transaction() as sess:
|
|
sess["authenticated"] = True
|
|
resp = self.client.get("/api/kiosk/setup-available")
|
|
self.assertTrue(resp.get_json().get("can_issue_access_link"))
|
|
|
|
def test_access_link_allowed_with_admin_session(self) -> None:
|
|
with self.client.session_transaction() as sess:
|
|
sess["authenticated"] = True
|
|
resp = self.client.get("/api/kiosk/access-link")
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertTrue(resp.get_json().get("start_url"))
|
|
|
|
def test_access_link_generates_permanent_slug(self) -> None:
|
|
first = self._access_link_payload()
|
|
second = self._access_link_payload()
|
|
self.assertTrue(first.get("permanent"))
|
|
self.assertEqual(first.get("start_url"), second.get("start_url"))
|
|
self.assertTrue(first["start_url"].endswith("/kiosk/start/" + first["start_url"].rsplit("/", 1)[-1]))
|
|
|
|
def test_access_link_refresh_rotates_slug(self) -> None:
|
|
first = self._access_link_payload()
|
|
old_slug = first["start_url"].rsplit("/", 1)[-1]
|
|
refreshed = self.client.post(
|
|
"/api/kiosk/access-link",
|
|
json={"refresh": True},
|
|
)
|
|
self.assertEqual(refreshed.status_code, 200)
|
|
new_url = refreshed.get_json().get("start_url") or ""
|
|
new_slug = new_url.rsplit("/", 1)[-1]
|
|
self.assertNotEqual(old_slug, new_slug)
|
|
stale = self._non_local_get(self.client, f"/kiosk/start/{old_slug}")
|
|
self.assertEqual(stale.status_code, 404)
|
|
|
|
def test_access_link_qr_image_is_local_svg_data_uri(self) -> None:
|
|
payload = self._access_link_payload()
|
|
uri = payload.get("qr_image_url") or ""
|
|
self.assertTrue(uri.startswith("data:image/svg+xml;base64,"))
|
|
raw = base64.b64decode(uri.split(",", 1)[1])
|
|
self.assertIn(b"<svg", raw.lower())
|
|
|
|
def test_start_url_sets_cookie_and_redirects(self) -> None:
|
|
resp = self._pair_via_start_url()
|
|
self.assertEqual(resp.status_code, 302)
|
|
self.assertIn("/scales", resp.headers.get("Location") or "")
|
|
self.assertIn("wesp_kiosk_device_id=", resp.headers.get("Set-Cookie") or "")
|
|
|
|
def test_fully_kiosk_flow_api_access(self) -> None:
|
|
fk = self.app.test_client()
|
|
start = self._pair_via_start_url(fk)
|
|
self.assertEqual(start.status_code, 302)
|
|
allowed = fk.get(
|
|
"/current_weight",
|
|
base_url="http://example.com",
|
|
environ_overrides={"REMOTE_ADDR": "10.0.0.55"},
|
|
)
|
|
self.assertEqual(allowed.status_code, 200)
|
|
|
|
def test_start_url_reopen_with_cookie(self) -> None:
|
|
fk = self.app.test_client()
|
|
slug = self._start_path_from_access_link()
|
|
first = fk.get(
|
|
f"/kiosk/start/{slug}",
|
|
base_url="http://example.com",
|
|
environ_overrides={"REMOTE_ADDR": "10.0.0.55"},
|
|
follow_redirects=False,
|
|
)
|
|
self.assertEqual(first.status_code, 302)
|
|
second = fk.get(
|
|
f"/kiosk/start/{slug}",
|
|
base_url="http://example.com",
|
|
environ_overrides={"REMOTE_ADDR": "10.0.0.55"},
|
|
follow_redirects=False,
|
|
)
|
|
self.assertEqual(second.status_code, 302)
|
|
|
|
def test_access_link_uses_configured_public_base_url(self) -> None:
|
|
self.app.config["KIOSK_PUBLIC_BASE_URL"] = "http://192.168.1.77:5000"
|
|
payload = self._access_link_payload()
|
|
self.assertTrue(payload["start_url"].startswith("http://192.168.1.77:5000/kiosk/start/"))
|
|
|
|
def test_access_link_localhost_uses_detected_lan_ip(self) -> None:
|
|
self.app.config["KIOSK_PUBLIC_BASE_URL"] = ""
|
|
with patch("app.routes.kiosk.detect_lan_ip", return_value="192.168.88.25"):
|
|
resp = self.client.get("/api/kiosk/access-link", base_url="http://localhost:5000")
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertTrue(
|
|
resp.get_json()["start_url"].startswith("http://192.168.88.25:5000/kiosk/start/")
|
|
)
|
|
|
|
def test_unpaired_then_start_url_then_api_access(self) -> None:
|
|
status = self.client.get("/api/kiosk/status")
|
|
self.assertFalse(status.get_json().get("paired"))
|
|
|
|
blocked = self.client.get(
|
|
"/current_weight",
|
|
base_url="http://example.com",
|
|
environ_overrides={"REMOTE_ADDR": "10.0.0.55"},
|
|
)
|
|
self.assertEqual(blocked.status_code, 401)
|
|
|
|
fk = self.app.test_client()
|
|
self._pair_via_start_url(fk)
|
|
status2 = fk.get(
|
|
"/api/kiosk/status",
|
|
base_url="http://example.com",
|
|
environ_overrides={"REMOTE_ADDR": "10.0.0.55"},
|
|
)
|
|
self.assertTrue(status2.get_json().get("paired"))
|
|
|
|
allowed = fk.get(
|
|
"/current_weight",
|
|
base_url="http://example.com",
|
|
environ_overrides={"REMOTE_ADDR": "10.0.0.55"},
|
|
)
|
|
self.assertEqual(allowed.status_code, 200)
|
|
|
|
def test_unpaired_then_pair_then_api_access(self) -> None:
|
|
status = self.client.get("/api/kiosk/status")
|
|
self.assertEqual(status.status_code, 200)
|
|
body0 = status.get_json()
|
|
self.assertFalse(body0.get("paired"))
|
|
self.assertTrue(body0.get("enforce_paired_only"))
|
|
|
|
blocked = self.client.get(
|
|
"/current_weight",
|
|
base_url="http://example.com",
|
|
environ_overrides={"REMOTE_ADDR": "10.0.0.55"},
|
|
)
|
|
self.assertEqual(blocked.status_code, 401)
|
|
|
|
token = self._pair()
|
|
status2 = self.client.get("/api/kiosk/status")
|
|
self.assertTrue(status2.get_json().get("paired"))
|
|
|
|
allowed = self.client.get("/current_weight")
|
|
self.assertEqual(allowed.status_code, 200)
|
|
|
|
reused = self.client.get(f"/api/kiosk/pair/confirm?token={token}")
|
|
self.assertEqual(reused.status_code, 200)
|
|
self.assertIn("уже", reused.get_data(as_text=True).lower())
|
|
|
|
again = self._confirm_post(token)
|
|
self.assertEqual(again.status_code, 409)
|
|
|
|
def test_pair_token_qr_image_is_local_svg_data_uri(self) -> None:
|
|
pair = self.client.post("/api/kiosk/pair-token")
|
|
self.assertEqual(pair.status_code, 200)
|
|
payload = pair.get_json()
|
|
uri = payload.get("qr_image_url") or ""
|
|
self.assertTrue(uri.startswith("data:image/svg+xml;base64,"), msg=repr(uri[:100]))
|
|
self.assertNotIn("qrserver", uri)
|
|
raw = base64.b64decode(uri.split(",", 1)[1])
|
|
self.assertIn(b"<svg", raw.lower())
|
|
|
|
def test_pair_token_single_use(self) -> None:
|
|
token = self._pair()
|
|
reused_get = self.client.get(f"/api/kiosk/pair/confirm?token={token}")
|
|
self.assertEqual(reused_get.status_code, 200)
|
|
self.assertIn("уже", reused_get.get_data(as_text=True).lower())
|
|
|
|
reused_post = self._confirm_post(token)
|
|
self.assertEqual(reused_post.status_code, 409)
|
|
|
|
def test_localhost_bypass_for_kiosk_pages(self) -> None:
|
|
kiosk_paths = [
|
|
"/scales",
|
|
"/calibration",
|
|
"/db_check",
|
|
"/duplicate",
|
|
"/unloading",
|
|
]
|
|
for path in kiosk_paths:
|
|
resp = self.client.get(path, base_url="http://localhost:5000")
|
|
self.assertEqual(resp.status_code, 200, msg=f"localhost should bypass guard for {path}")
|
|
|
|
def test_localhost_bypass_for_protected_api(self) -> None:
|
|
weight_resp = self.client.get("/current_weight", base_url="http://localhost:5000")
|
|
self.assertEqual(weight_resp.status_code, 200)
|
|
sse_resp = self.client.get("/stream_weight?r=test", base_url="http://localhost:5000")
|
|
self.assertEqual(sse_resp.status_code, 200)
|
|
|
|
def test_unpaired_non_localhost_redirects_to_unauthorized(self) -> None:
|
|
external_client = self.app.test_client()
|
|
resp = self._non_local_get(external_client, "/scales")
|
|
self.assertEqual(resp.status_code, 302)
|
|
location = resp.headers.get("Location") or ""
|
|
self.assertIn("/unauthorized-device", location)
|
|
self.assertTrue(
|
|
"next=%2Fscales" in location or "next=/scales" in location,
|
|
msg=location,
|
|
)
|
|
self.assertIn("wesp_kiosk_device_id=", (resp.headers.get("Set-Cookie") or ""))
|
|
|
|
def test_confirm_post_activates_confirming_browser_when_different_from_token_device(
|
|
self,
|
|
) -> None:
|
|
"""Главный (localhost) выпускает токен; телефон с другим cookie подтверждает — оба active."""
|
|
source_terminal = self.client
|
|
external_client = self.app.test_client()
|
|
|
|
external_resp = self._non_local_get(external_client, "/scales")
|
|
self.assertEqual(external_resp.status_code, 302)
|
|
|
|
pair = source_terminal.post("/api/kiosk/pair-token")
|
|
self.assertEqual(pair.status_code, 200)
|
|
token = pair.get_json().get("token")
|
|
self.assertTrue(token)
|
|
|
|
ok = external_client.post(
|
|
"/api/kiosk/pair/confirm",
|
|
data={"token": token},
|
|
content_type="application/x-www-form-urlencoded",
|
|
base_url="http://example.com",
|
|
environ_overrides={"REMOTE_ADDR": "10.0.0.55"},
|
|
)
|
|
self.assertEqual(ok.status_code, 302)
|
|
|
|
allowed = external_client.get(
|
|
"/current_weight",
|
|
base_url="http://example.com",
|
|
environ_overrides={"REMOTE_ADDR": "10.0.0.55"},
|
|
)
|
|
self.assertEqual(allowed.status_code, 200)
|
|
|
|
def test_scan_confirm_pairs_external_device(self) -> None:
|
|
source_terminal = self.client
|
|
external_client = self.app.test_client()
|
|
|
|
external_resp = self._non_local_get(external_client, "/scales")
|
|
self.assertEqual(external_resp.status_code, 302)
|
|
|
|
pair = source_terminal.post("/api/kiosk/pair-token")
|
|
self.assertEqual(pair.status_code, 200)
|
|
token = pair.get_json().get("token")
|
|
self.assertTrue(token)
|
|
|
|
confirm = external_client.post(
|
|
"/api/kiosk/pair/scan-confirm",
|
|
json={"token": token},
|
|
base_url="http://example.com",
|
|
environ_overrides={"REMOTE_ADDR": "10.0.0.55"},
|
|
)
|
|
self.assertEqual(confirm.status_code, 200)
|
|
self.assertTrue(confirm.get_json().get("paired"))
|
|
|
|
allowed = external_client.get(
|
|
"/current_weight",
|
|
base_url="http://example.com",
|
|
environ_overrides={"REMOTE_ADDR": "10.0.0.55"},
|
|
)
|
|
self.assertEqual(allowed.status_code, 200)
|
|
|
|
def test_claim_link_pairs_external_device(self) -> None:
|
|
source_terminal = self.client
|
|
external_client = self.app.test_client()
|
|
|
|
external_resp = self._non_local_get(external_client, "/scales")
|
|
self.assertEqual(external_resp.status_code, 302)
|
|
|
|
pair = source_terminal.post("/api/kiosk/pair-token")
|
|
self.assertEqual(pair.status_code, 200)
|
|
token = pair.get_json().get("token")
|
|
self.assertTrue(token)
|
|
|
|
claim = external_client.get(
|
|
f"/api/kiosk/pair/claim?token={token}",
|
|
base_url="http://example.com",
|
|
environ_overrides={"REMOTE_ADDR": "10.0.0.55"},
|
|
)
|
|
self.assertEqual(claim.status_code, 200)
|
|
|
|
allowed = external_client.get(
|
|
"/current_weight",
|
|
base_url="http://example.com",
|
|
environ_overrides={"REMOTE_ADDR": "10.0.0.55"},
|
|
)
|
|
self.assertEqual(allowed.status_code, 200)
|
|
|
|
def test_pair_token_uses_configured_public_base_url(self) -> None:
|
|
self.app.config["KIOSK_PUBLIC_BASE_URL"] = "http://192.168.1.77:5000"
|
|
resp = self.client.post("/api/kiosk/pair-token", base_url="http://localhost:5000")
|
|
self.assertEqual(resp.status_code, 200)
|
|
payload = resp.get_json()
|
|
self.assertTrue(payload["pair_url"].startswith("http://192.168.1.77:5000/api/kiosk/pair/confirm?token="))
|
|
self.assertTrue(payload["confirm_url"].startswith("http://192.168.1.77:5000/api/kiosk/pair/confirm?token="))
|
|
|
|
def test_pair_token_localhost_uses_detected_lan_ip(self) -> None:
|
|
self.app.config["KIOSK_PUBLIC_BASE_URL"] = ""
|
|
with patch("app.routes.kiosk.detect_lan_ip", return_value="192.168.88.25"):
|
|
resp = self.client.post("/api/kiosk/pair-token", base_url="http://localhost:5000")
|
|
self.assertEqual(resp.status_code, 200)
|
|
payload = resp.get_json()
|
|
self.assertTrue(payload["pair_url"].startswith("http://192.168.88.25:5000/api/kiosk/pair/confirm?token="))
|
|
self.assertTrue(payload["confirm_url"].startswith("http://192.168.88.25:5000/api/kiosk/pair/confirm?token="))
|
|
|
|
def test_pair_token_localhost_falls_back_to_request_host(self) -> None:
|
|
self.app.config["KIOSK_PUBLIC_BASE_URL"] = ""
|
|
with patch("app.routes.kiosk.detect_lan_ip", return_value=""):
|
|
resp = self.client.post("/api/kiosk/pair-token", base_url="http://localhost:5000")
|
|
self.assertEqual(resp.status_code, 200)
|
|
payload = resp.get_json()
|
|
self.assertTrue(payload["pair_url"].startswith("http://localhost:5000/api/kiosk/pair/confirm?token="))
|
|
self.assertTrue(payload["confirm_url"].startswith("http://localhost:5000/api/kiosk/pair/confirm?token="))
|
|
|
|
|
|
class KioskEnforceOffConfig(KioskPairConfig):
|
|
KIOSK_ENFORCE_PAIRED_ONLY = False
|
|
|
|
|
|
class KioskPairingEnforceOffTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.app = create_app(KioskEnforceOffConfig)
|
|
self.client = self.app.test_client()
|
|
self.ctx = self.app.app_context()
|
|
self.ctx.push()
|
|
db.create_all()
|
|
self.client.get("/scales")
|
|
|
|
def tearDown(self) -> None:
|
|
from app.routes import scales as scales_module
|
|
|
|
reader = getattr(scales_module, "_scales_reader", None)
|
|
if reader is not None:
|
|
reader.stop()
|
|
scales_module._scales_reader = None
|
|
db.session.remove()
|
|
db.drop_all()
|
|
self.ctx.pop()
|
|
|
|
def test_kiosk_status_includes_enforce_off(self) -> None:
|
|
st = self.client.get("/api/kiosk/status")
|
|
self.assertEqual(st.status_code, 200)
|
|
body = st.get_json()
|
|
self.assertIn("enforce_paired_only", body)
|
|
self.assertFalse(body.get("enforce_paired_only"))
|
|
|
|
def test_scales_page_not_redirected_when_enforce_off(self) -> None:
|
|
ext = self.app.test_client()
|
|
resp = ext.get(
|
|
"/scales",
|
|
base_url="http://kiosk.example:5000",
|
|
environ_overrides={"REMOTE_ADDR": "10.0.0.99"},
|
|
follow_redirects=False,
|
|
)
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|