test(kiosk): E2E-Test für Login + Stempel-Flow, Cross-Device-Session-Check

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Gis16MnuwkYcivLrSxK1pD

backend/tests/test_kiosk_stamp.py

@@ -0,0 +1,147 @@
+"""Tests für Kiosk-Stempel-Endpunkte (Login -> stamp/in|out|status)."""
+import base64
+import hashlib
+import time
+import uuid
+
+import pytest
+import pytest_asyncio
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
+from httpx import AsyncClient
+
+pytestmark = pytest.mark.asyncio
+
+DEVICES_URL = "/api/v1/kiosk/devices"
+
+
+def _headers(device_id: str, private_key: Ed25519PrivateKey, path: str, body: bytes) -> dict:
+    timestamp = str(int(time.time()))
+    nonce = str(uuid.uuid4())
+    body_hash = hashlib.sha256(body).hexdigest()
+    message = f"POST {path} {timestamp} {nonce} {body_hash}".encode()
+    signature = private_key.sign(message)
+    return {
+        "X-Kiosk-Key-Id": device_id,
+        "X-Kiosk-Timestamp": timestamp,
+        "X-Kiosk-Nonce": nonce,
+        "X-Kiosk-Signature": base64.b64encode(signature).decode(),
+        "Content-Type": "application/json",
+    }
+
+
+@pytest_asyncio.fixture(scope="session", loop_scope="session")
+async def stamp_keypair():
+    private_key = Ed25519PrivateKey.generate()
+    public_key_pem = (
+        private_key.public_key()
+        .public_bytes(Encoding.PEM, PublicFormat.SubjectPublicKeyInfo)
+        .decode()
+    )
+    return private_key, public_key_pem
+
+
+@pytest_asyncio.fixture(scope="session", loop_scope="session")
+async def stamp_user_headers(client: AsyncClient):
+    resp = await client.post("/api/v1/auth/register", json={
+        "company_name": "Stamp GmbH", "first_name": "Stan", "last_name": "Pel",
+        "email": "admin@stampgmbh.de", "password": "Secret123",
+    })
+    assert resp.status_code == 201, resp.text
+    return {"Authorization": f"Bearer {resp.json()['access_token']}"}
+
+
+@pytest_asyncio.fixture(scope="session", loop_scope="session")
+async def stamp_device(client: AsyncClient, stamp_user_headers, stamp_keypair):
+    _, public_key_pem = stamp_keypair
+    resp = await client.post(DEVICES_URL,
+        json={"name": "Stamp-Terminal", "location": "Test", "public_key": public_key_pem},
+        headers=stamp_user_headers)
+    assert resp.status_code == 201, resp.text
+    device = resp.json()
+    ap = await client.post(f"{DEVICES_URL}/{device['id']}/approve", headers=stamp_user_headers)
+    assert ap.status_code == 200, ap.text
+    return device
+
+
+async def test_pin_login_then_stamp_in_and_out(
+    client: AsyncClient, stamp_user_headers, stamp_keypair, stamp_device
+):
+    private_key, _ = stamp_keypair
+    device_id = stamp_device["id"]
+
+    me = await client.get("/api/v1/users/me", headers=stamp_user_headers)
+    user_id = me.json()["id"]
+    personnel_number = me.json().get("personnel_number")
+    if not personnel_number:
+        pn = await client.patch(f"/api/v1/users/{user_id}",
+                                json={"personnel_number": "9001"}, headers=stamp_user_headers)
+        assert pn.status_code == 200, pn.text
+        personnel_number = "9001"
+
+    pin_r = await client.post(f"/api/v1/users/{user_id}/kiosk-pin",
+                              json={"pin": "4321"}, headers=stamp_user_headers)
+    assert pin_r.status_code == 200, pin_r.text
+
+    login_body = f'{{"personnel_number": "{personnel_number}", "pin": "4321"}}'.encode()
+    login_resp = await client.post(
+        "/api/v1/kiosk/auth/pin", content=login_body,
+        headers=_headers(device_id, private_key, "/api/v1/kiosk/auth/pin", login_body),
+    )
+    assert login_resp.status_code == 200, login_resp.text
+    session_token = login_resp.json()["session_token"]
+
+    status_body = f'{{"session_token": "{session_token}"}}'.encode()
+    status_resp = await client.post(
+        "/api/v1/kiosk/stamp/status", content=status_body,
+        headers=_headers(device_id, private_key, "/api/v1/kiosk/stamp/status", status_body),
+    )
+    assert status_resp.status_code == 200, status_resp.text
+    assert status_resp.json()["stamped_in"] is False
+
+    in_body = f'{{"session_token": "{session_token}"}}'.encode()
+    in_resp = await client.post(
+        "/api/v1/kiosk/stamp/in", content=in_body,
+        headers=_headers(device_id, private_key, "/api/v1/kiosk/stamp/in", in_body),
+    )
+    assert in_resp.status_code == 200, in_resp.text
+    assert in_resp.json()["status"] == "stamped_in"
+
+    out_body = f'{{"session_token": "{session_token}"}}'.encode()
+    out_resp = await client.post(
+        "/api/v1/kiosk/stamp/out", content=out_body,
+        headers=_headers(device_id, private_key, "/api/v1/kiosk/stamp/out", out_body),
+    )
+    assert out_resp.status_code == 200, out_resp.text
+    assert out_resp.json()["status"] == "stamped_out"
+
+
+async def test_stamp_rejects_session_from_other_device(
+    client: AsyncClient, stamp_user_headers, stamp_keypair, stamp_device
+):
+    private_key, public_key_pem = stamp_keypair
+    other = await client.post(DEVICES_URL,
+        json={"name": "Anderes-Terminal", "location": "Test", "public_key": public_key_pem},
+        headers=stamp_user_headers)
+    assert other.status_code == 201
+    other_device = other.json()
+    ap = await client.post(f"{DEVICES_URL}/{other_device['id']}/approve", headers=stamp_user_headers)
+    assert ap.status_code == 200
+
+    me = await client.get("/api/v1/users/me", headers=stamp_user_headers)
+    user_id = me.json()["id"]
+    login_body = f'{{"user_id": "{user_id}"}}'.encode()
+    login_resp = await client.post(
+        "/api/v1/kiosk/auth/list", content=login_body,
+        headers=_headers(stamp_device["id"], private_key, "/api/v1/kiosk/auth/list", login_body),
+    )
+    assert login_resp.status_code == 200, login_resp.text
+    session_token = login_resp.json()["session_token"]
+
+    # Session wurde für stamp_device erstellt - Stempeln über other_device muss scheitern
+    body = f'{{"session_token": "{session_token}"}}'.encode()
+    resp = await client.post(
+        "/api/v1/kiosk/stamp/in", content=body,
+        headers=_headers(other_device["id"], private_key, "/api/v1/kiosk/stamp/in", body),
+    )
+    assert resp.status_code == 401, resp.text