diff --git a/backend/tests/test_kiosk_stamp.py b/backend/tests/test_kiosk_stamp.py new file mode 100644 index 0000000..ba0e755 --- /dev/null +++ b/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