Samir Boulahtit
3bf23c1b23
Some checks failed
Two-pane landscape: scrollable staff list on the left, PIN dots + numeric
keypad on the right. Footer shows online/offline + pending-sync count.
Going with cached-hashes for offline-capable PIN verify (decision logged
in chat). The threat model already accepts the device — a stolen tablet
holds a 1-year store-scoped JWT, so leaking 4-digit bcrypt hashes is
incremental. Hashes only ever leave the server when the requester is a
paired POS tablet, gated by the new endpoint refusing user JWTs.
Backend:
- GET /api/v1/store/loyalty/pins/for-device — returns PINs WITH pin_hash
for terminal-device JWTs only; user JWTs receive 403.
- PinForDeviceResponse / PinForDeviceListResponse schemas.
- 2 integration tests in TestPinsForDevice (10/10 pass total).
Android:
- PinForDeviceItem / PinForDeviceListResponse Moshi models.
- LoyaltyApi.listPinsForDevice().
- StaffPinRepository.verifyPin(plain) — at.favre.lib bcrypt verify
against cached hashes; filters active + unlocked rows in one pass.
- PendingTransactionDao.getPendingCount() switched to Flow<Int> so the
badge auto-updates when transactions sync.
- PinViewModel state machine — loads pins on init, accumulates digits,
bcrypt-verifies on length >= 4, fires verified/errorMessage. Combines
pending-sync count + online state into the same StateFlow.
- PinScreen rewrite: avatar-circle staff list, 6-dot PIN display,
spinner during verify, error label on wrong PIN, status footer.
Open follow-up (intentional, post-launch): tablet doesn't yet report
failed attempts back to the server's lockout counter. Path is clear —
small POST /pins/{id}/record-failed-attempt endpoint plus a call from
attemptVerify's failure branch.
Verified by ./gradlew assembleDebug — clean build.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
312 lines
10 KiB
Python
312 lines
10 KiB
Python
# app/modules/loyalty/tests/integration/test_terminal_devices.py
|
|
"""Integration tests for terminal device pairing & revocation.
|
|
|
|
Covers both the merchant API (`/api/v1/merchants/loyalty/devices/...`) and
|
|
the admin-on-behalf API (`/api/v1/admin/loyalty/merchants/{id}/devices/...`),
|
|
plus the bearer-auth path on the store API that accepts a paired-device JWT.
|
|
"""
|
|
|
|
import pytest
|
|
from jose import jwt
|
|
|
|
from app.modules.loyalty.models import TerminalDevice
|
|
from middleware.auth import AuthManager
|
|
|
|
MERCHANT_BASE = "/api/v1/merchants/loyalty"
|
|
STORE_BASE = "/api/v1/store/loyalty"
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.api
|
|
@pytest.mark.loyalty
|
|
class TestMerchantPairing:
|
|
"""POST /merchants/loyalty/devices — creates row and returns one-time QR."""
|
|
|
|
def test_pair_device_returns_setup_token_and_qr(
|
|
self, client, loyalty_merchant_headers, loyalty_store_setup, db
|
|
):
|
|
store = loyalty_store_setup["store"]
|
|
response = client.post(
|
|
f"{MERCHANT_BASE}/devices",
|
|
json={"store_id": store.id, "label": "Counter 1"},
|
|
headers=loyalty_merchant_headers,
|
|
)
|
|
assert response.status_code == 201, response.text
|
|
data = response.json()
|
|
|
|
assert data["label"] == "Counter 1"
|
|
assert data["store_id"] == store.id
|
|
assert data["status"] == "active"
|
|
assert data["setup_token"]
|
|
assert data["qr_png_base64"].startswith("data:image/png;base64,")
|
|
assert data["setup_payload"]["store_code"] == store.store_code
|
|
assert data["setup_payload"]["auth_token"] == data["setup_token"]
|
|
|
|
# Token is signed by AuthManager and carries the device_setup claim
|
|
# plus the same jti as the row.
|
|
auth = AuthManager()
|
|
decoded = jwt.decode(
|
|
data["setup_token"], auth.secret_key, algorithms=[auth.algorithm]
|
|
)
|
|
assert decoded["device_setup"] is True
|
|
assert decoded["store_id"] == store.id
|
|
device = (
|
|
db.query(TerminalDevice)
|
|
.filter(TerminalDevice.id == data["id"])
|
|
.first()
|
|
)
|
|
assert device is not None
|
|
assert device.jti == decoded["jti"]
|
|
|
|
def test_pair_device_rejects_foreign_store(
|
|
self, client, loyalty_merchant_headers, db
|
|
):
|
|
"""A merchant cannot pair against a store they don't own."""
|
|
# store_id=99999 doesn't belong to this merchant
|
|
response = client.post(
|
|
f"{MERCHANT_BASE}/devices",
|
|
json={"store_id": 99999, "label": "Hijack"},
|
|
headers=loyalty_merchant_headers,
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.api
|
|
@pytest.mark.loyalty
|
|
class TestMerchantListAndRevoke:
|
|
"""GET / POST revoke — listing and lifecycle."""
|
|
|
|
def test_list_excludes_revoked_by_default(
|
|
self, client, loyalty_merchant_headers, loyalty_store_setup
|
|
):
|
|
store = loyalty_store_setup["store"]
|
|
# Pair two devices
|
|
for label in ("A", "B"):
|
|
r = client.post(
|
|
f"{MERCHANT_BASE}/devices",
|
|
json={"store_id": store.id, "label": label},
|
|
headers=loyalty_merchant_headers,
|
|
)
|
|
assert r.status_code == 201
|
|
# Revoke first one
|
|
listing = client.get(
|
|
f"{MERCHANT_BASE}/devices", headers=loyalty_merchant_headers
|
|
).json()
|
|
first_id = listing["devices"][-1]["id"] # deterministic order: created_at desc
|
|
r = client.post(
|
|
f"{MERCHANT_BASE}/devices/{first_id}/revoke",
|
|
json={"reason": "lost"},
|
|
headers=loyalty_merchant_headers,
|
|
)
|
|
assert r.status_code == 200
|
|
assert r.json()["status"] == "revoked"
|
|
|
|
# Default list excludes the revoked one
|
|
active = client.get(
|
|
f"{MERCHANT_BASE}/devices", headers=loyalty_merchant_headers
|
|
).json()
|
|
assert active["total"] == 1
|
|
|
|
# include_revoked=true brings it back
|
|
full = client.get(
|
|
f"{MERCHANT_BASE}/devices?include_revoked=true",
|
|
headers=loyalty_merchant_headers,
|
|
).json()
|
|
assert full["total"] == 2
|
|
|
|
def test_revoke_is_idempotent(
|
|
self, client, loyalty_merchant_headers, loyalty_store_setup
|
|
):
|
|
store = loyalty_store_setup["store"]
|
|
r = client.post(
|
|
f"{MERCHANT_BASE}/devices",
|
|
json={"store_id": store.id, "label": "Idem"},
|
|
headers=loyalty_merchant_headers,
|
|
).json()
|
|
first = client.post(
|
|
f"{MERCHANT_BASE}/devices/{r['id']}/revoke",
|
|
json={},
|
|
headers=loyalty_merchant_headers,
|
|
).json()
|
|
second = client.post(
|
|
f"{MERCHANT_BASE}/devices/{r['id']}/revoke",
|
|
json={},
|
|
headers=loyalty_merchant_headers,
|
|
).json()
|
|
assert first["revoked_at"] == second["revoked_at"]
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.api
|
|
@pytest.mark.loyalty
|
|
class TestStoreAPIBearerAuth:
|
|
"""Paired-device JWT is accepted on the store API."""
|
|
|
|
def test_device_token_authenticates_store_api(
|
|
self, client, loyalty_merchant_headers, loyalty_store_setup
|
|
):
|
|
store = loyalty_store_setup["store"]
|
|
paired = client.post(
|
|
f"{MERCHANT_BASE}/devices",
|
|
json={"store_id": store.id, "label": "Counter 1"},
|
|
headers=loyalty_merchant_headers,
|
|
).json()
|
|
token = paired["setup_token"]
|
|
|
|
response = client.get(
|
|
f"{STORE_BASE}/program",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
body = response.json()
|
|
assert body["merchant_id"] == loyalty_store_setup["merchant"].id
|
|
|
|
def test_revoked_device_token_is_rejected(
|
|
self, client, loyalty_merchant_headers, loyalty_store_setup
|
|
):
|
|
store = loyalty_store_setup["store"]
|
|
paired = client.post(
|
|
f"{MERCHANT_BASE}/devices",
|
|
json={"store_id": store.id, "label": "Burn"},
|
|
headers=loyalty_merchant_headers,
|
|
).json()
|
|
token = paired["setup_token"]
|
|
|
|
client.post(
|
|
f"{MERCHANT_BASE}/devices/{paired['id']}/revoke",
|
|
json={"reason": "test"},
|
|
headers=loyalty_merchant_headers,
|
|
)
|
|
|
|
response = client.get(
|
|
f"{STORE_BASE}/program",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert response.status_code in (400, 401)
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.api
|
|
@pytest.mark.loyalty
|
|
class TestActingDeviceAudit:
|
|
"""Transactions performed via a device JWT carry acting_terminal_device_id."""
|
|
|
|
def test_points_earn_via_device_stamps_audit_column(
|
|
self, client, loyalty_merchant_headers, loyalty_store_setup, db
|
|
):
|
|
from app.modules.loyalty.models import LoyaltyTransaction
|
|
from app.modules.loyalty.models.loyalty_transaction import TransactionType
|
|
|
|
store = loyalty_store_setup["store"]
|
|
card = loyalty_store_setup["card"]
|
|
|
|
paired = client.post(
|
|
f"{MERCHANT_BASE}/devices",
|
|
json={"store_id": store.id, "label": "Counter 1"},
|
|
headers=loyalty_merchant_headers,
|
|
).json()
|
|
device_id = paired["id"]
|
|
token = paired["setup_token"]
|
|
|
|
response = client.post(
|
|
f"{STORE_BASE}/points/earn",
|
|
json={"card_id": card.id, "purchase_amount_cents": 1500},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
|
|
tx = (
|
|
db.query(LoyaltyTransaction)
|
|
.filter(
|
|
LoyaltyTransaction.card_id == card.id,
|
|
LoyaltyTransaction.transaction_type
|
|
== TransactionType.POINTS_EARNED.value,
|
|
)
|
|
.order_by(LoyaltyTransaction.id.desc())
|
|
.first()
|
|
)
|
|
assert tx is not None
|
|
assert tx.acting_terminal_device_id == device_id
|
|
|
|
def test_user_token_leaves_audit_column_null(
|
|
self, client, loyalty_store_headers, loyalty_store_setup, db
|
|
):
|
|
"""Web-terminal user JWT must NOT stamp acting_terminal_device_id."""
|
|
from app.modules.loyalty.models import LoyaltyTransaction
|
|
from app.modules.loyalty.models.loyalty_transaction import TransactionType
|
|
|
|
card = loyalty_store_setup["card"]
|
|
|
|
response = client.post(
|
|
f"{STORE_BASE}/points/earn",
|
|
json={"card_id": card.id, "purchase_amount_cents": 1500},
|
|
headers=loyalty_store_headers,
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
|
|
tx = (
|
|
db.query(LoyaltyTransaction)
|
|
.filter(
|
|
LoyaltyTransaction.card_id == card.id,
|
|
LoyaltyTransaction.transaction_type
|
|
== TransactionType.POINTS_EARNED.value,
|
|
)
|
|
.order_by(LoyaltyTransaction.id.desc())
|
|
.first()
|
|
)
|
|
assert tx is not None
|
|
assert tx.acting_terminal_device_id is None
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.api
|
|
@pytest.mark.loyalty
|
|
class TestPinsForDevice:
|
|
"""GET /pins/for-device exposes hashes only to paired devices."""
|
|
|
|
def test_device_token_receives_hashes(
|
|
self, client, loyalty_merchant_headers, loyalty_store_setup, db
|
|
):
|
|
from app.modules.loyalty.models import StaffPin
|
|
|
|
store = loyalty_store_setup["store"]
|
|
program = loyalty_store_setup["program"]
|
|
pin = StaffPin(
|
|
merchant_id=store.merchant_id,
|
|
program_id=program.id,
|
|
store_id=store.id,
|
|
name="Test Cashier",
|
|
staff_id="cash01",
|
|
)
|
|
pin.set_pin("4321")
|
|
db.add(pin)
|
|
db.commit()
|
|
db.refresh(pin)
|
|
|
|
paired = client.post(
|
|
f"{MERCHANT_BASE}/devices",
|
|
json={"store_id": store.id, "label": "PIN test"},
|
|
headers=loyalty_merchant_headers,
|
|
).json()
|
|
token = paired["setup_token"]
|
|
|
|
response = client.get(
|
|
f"{STORE_BASE}/pins/for-device",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
payload = response.json()
|
|
assert payload["total"] >= 1
|
|
target = next(p for p in payload["pins"] if p["id"] == pin.id)
|
|
assert target["pin_hash"].startswith("$2") and len(target["pin_hash"]) > 50
|
|
|
|
def test_user_token_is_rejected(
|
|
self, client, loyalty_store_headers
|
|
):
|
|
response = client.get(
|
|
f"{STORE_BASE}/pins/for-device",
|
|
headers=loyalty_store_headers,
|
|
)
|
|
assert response.status_code == 403, response.text
|