feat(loyalty): production readiness round 2 — 12 security, integrity & correctness fixes
Some checks failed
CI / ruff (push) Successful in 12s
CI / validate (push) Successful in 27s
CI / dependency-scanning (push) Successful in 31s
CI / pytest (push) Failing after 3h14m58s
CI / docs (push) Has been cancelled
CI / deploy (push) Has been cancelled

Security:
- Fix TOCTOU race conditions: move balance/limit checks after row lock in redeem_points, add_stamp, redeem_stamps
- Add PIN ownership verification to update/delete/unlock store routes
- Gate adjust_points endpoint to merchant_owner role only

Data integrity:
- Track total_points_voided in void_points
- Add order_reference idempotency guard in earn_points

Correctness:
- Fix LoyaltyProgramAlreadyExistsException to use merchant_id parameter
- Add StorefrontProgramResponse excluding wallet IDs from public API
- Add bounds (±100000) to PointsAdjustRequest.points_delta

Audit & config:
- Add CARD_REACTIVATED transaction type with audit record
- Improve admin audit logging with actor identity and old values
- Use merchant-specific PIN lockout settings with global fallback
- Guard MerchantLoyaltySettings creation with get_or_create pattern

Tests: 27 new tests (265 total) covering all 12 items — unit and integration.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-16 23:37:23 +01:00
parent b6047f5b7d
commit 7d652716bb
20 changed files with 955 additions and 28 deletions

View File

@@ -462,3 +462,343 @@ class TestStampEarnRedeem:
data = response.json()
assert data["success"] is True
assert data["stamp_count"] == 0
# ============================================================================
# Item 2: PIN Ownership Checks
# ============================================================================
@pytest.fixture
def pin_ownership_setup(db, loyalty_platform):
"""Setup two stores under same merchant for PIN ownership tests."""
from app.modules.customers.models.customer import Customer
from app.modules.loyalty.models import StaffPin
from app.modules.tenancy.models.store import StoreUser
from app.modules.tenancy.models.store_platform import StorePlatform
from middleware.auth import AuthManager
auth = AuthManager()
uid = uuid.uuid4().hex[:8]
owner = User(
email=f"pinown_{uid}@test.com",
username=f"pinown_{uid}",
hashed_password=auth.hash_password("storepass123"),
role="merchant_owner",
is_active=True,
is_email_verified=True,
)
db.add(owner)
db.commit()
db.refresh(owner)
from app.modules.tenancy.models import Merchant, Store
merchant = Merchant(
name=f"Pin Own Merchant {uid}",
owner_user_id=owner.id,
contact_email=owner.email,
is_active=True,
is_verified=True,
)
db.add(merchant)
db.commit()
db.refresh(merchant)
# Store A (owner's store)
store_a = Store(
merchant_id=merchant.id,
store_code=f"PINA_{uid.upper()}",
subdomain=f"pina{uid}",
name=f"Pin Store A {uid}",
is_active=True,
is_verified=True,
)
db.add(store_a)
db.commit()
db.refresh(store_a)
store_user = StoreUser(store_id=store_a.id, user_id=owner.id, is_active=True)
db.add(store_user)
db.commit()
sp = StorePlatform(store_id=store_a.id, platform_id=loyalty_platform.id)
db.add(sp)
db.commit()
# Store B (different store)
store_b = Store(
merchant_id=merchant.id,
store_code=f"PINB_{uid.upper()}",
subdomain=f"pinb{uid}",
name=f"Pin Store B {uid}",
is_active=True,
is_verified=True,
)
db.add(store_b)
db.commit()
db.refresh(store_b)
program = LoyaltyProgram(
merchant_id=merchant.id,
loyalty_type=LoyaltyType.POINTS.value,
points_per_euro=10,
cooldown_minutes=0,
max_daily_stamps=10,
require_staff_pin=True,
card_name="Pin Own Card",
card_color="#00FF00",
is_active=True,
)
db.add(program)
db.commit()
db.refresh(program)
# Create PIN on store B
pin_b = StaffPin(
merchant_id=merchant.id,
program_id=program.id,
store_id=store_b.id,
name="Other Store Staff",
)
pin_b.set_pin("5678")
db.add(pin_b)
db.commit()
db.refresh(pin_b)
return {
"owner": owner,
"merchant": merchant,
"store_a": store_a,
"store_b": store_b,
"program": program,
"pin_b": pin_b,
}
@pytest.fixture
def pin_ownership_headers(client, pin_ownership_setup):
"""JWT auth headers for pin ownership setup (logged into store A)."""
owner = pin_ownership_setup["owner"]
response = client.post(
"/api/v1/store/auth/login",
json={"email_or_username": owner.username, "password": "storepass123"},
)
assert response.status_code == 200
token = response.json()["access_token"]
return {"Authorization": f"Bearer {token}"}
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.loyalty
class TestPinOwnershipCheck:
"""Tests for PIN ownership verification (Item 2)."""
def test_update_pin_wrong_store_returns_404(
self, client, pin_ownership_headers, pin_ownership_setup
):
"""Updating a PIN belonging to another store returns 404."""
pin_b = pin_ownership_setup["pin_b"]
response = client.patch(
f"{BASE}/pins/{pin_b.id}",
json={"name": "Hacked Name"},
headers=pin_ownership_headers,
)
assert response.status_code == 404
def test_delete_pin_wrong_store_returns_404(
self, client, pin_ownership_headers, pin_ownership_setup
):
"""Deleting a PIN belonging to another store returns 404."""
pin_b = pin_ownership_setup["pin_b"]
response = client.delete(
f"{BASE}/pins/{pin_b.id}",
headers=pin_ownership_headers,
)
assert response.status_code == 404
def test_unlock_pin_wrong_store_returns_404(
self, client, pin_ownership_headers, pin_ownership_setup
):
"""Unlocking a PIN belonging to another store returns 404."""
pin_b = pin_ownership_setup["pin_b"]
response = client.post(
f"{BASE}/pins/{pin_b.id}/unlock",
headers=pin_ownership_headers,
)
assert response.status_code == 404
# ============================================================================
# Item 3: adjust_points role gate
# ============================================================================
@pytest.fixture
def staff_user_setup(db, loyalty_platform):
"""Setup with a staff user (not merchant_owner) for role gate tests."""
from app.modules.customers.models.customer import Customer
from app.modules.tenancy.models.store import StoreUser
from app.modules.tenancy.models.store_platform import StorePlatform
from middleware.auth import AuthManager
auth = AuthManager()
uid = uuid.uuid4().hex[:8]
# merchant_owner creates the merchant
owner = User(
email=f"adjowner_{uid}@test.com",
username=f"adjowner_{uid}",
hashed_password=auth.hash_password("storepass123"),
role="merchant_owner",
is_active=True,
is_email_verified=True,
)
db.add(owner)
db.commit()
db.refresh(owner)
from app.modules.tenancy.models import Merchant, Store
merchant = Merchant(
name=f"Adjust Merchant {uid}",
owner_user_id=owner.id,
contact_email=owner.email,
is_active=True,
is_verified=True,
)
db.add(merchant)
db.commit()
db.refresh(merchant)
store = Store(
merchant_id=merchant.id,
store_code=f"ADJ_{uid.upper()}",
subdomain=f"adj{uid}",
name=f"Adjust Store {uid}",
is_active=True,
is_verified=True,
)
db.add(store)
db.commit()
db.refresh(store)
# Staff user (not merchant_owner)
staff = User(
email=f"adjstaff_{uid}@test.com",
username=f"adjstaff_{uid}",
hashed_password=auth.hash_password("storepass123"),
role="store_member",
is_active=True,
is_email_verified=True,
)
db.add(staff)
db.commit()
db.refresh(staff)
store_user = StoreUser(store_id=store.id, user_id=staff.id, is_active=True)
db.add(store_user)
db.commit()
sp = StorePlatform(store_id=store.id, platform_id=loyalty_platform.id)
db.add(sp)
db.commit()
customer = Customer(
email=f"adjcust_{uid}@test.com",
first_name="Adj",
last_name="Customer",
hashed_password="!unused!", # noqa: SEC001
customer_number=f"AC-{uid.upper()}",
store_id=store.id,
is_active=True,
)
db.add(customer)
db.commit()
db.refresh(customer)
program = LoyaltyProgram(
merchant_id=merchant.id,
loyalty_type=LoyaltyType.POINTS.value,
points_per_euro=10,
cooldown_minutes=0,
max_daily_stamps=10,
require_staff_pin=False,
card_name="Adjust Card",
card_color="#FF0000",
is_active=True,
)
db.add(program)
db.commit()
db.refresh(program)
card = LoyaltyCard(
merchant_id=merchant.id,
program_id=program.id,
customer_id=customer.id,
enrolled_at_store_id=store.id,
card_number=f"ADJCARD-{uid.upper()}",
points_balance=500,
total_points_earned=500,
is_active=True,
last_activity_at=datetime.now(UTC),
)
db.add(card)
db.commit()
db.refresh(card)
return {
"staff": staff,
"merchant": merchant,
"store": store,
"program": program,
"card": card,
}
@pytest.fixture
def staff_headers(client, staff_user_setup):
"""JWT auth headers for staff user (not merchant_owner)."""
staff = staff_user_setup["staff"]
response = client.post(
"/api/v1/store/auth/login",
json={"email_or_username": staff.username, "password": "storepass123"},
)
assert response.status_code == 200
token = response.json()["access_token"]
return {"Authorization": f"Bearer {token}"}
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.loyalty
class TestAdjustPointsRoleGate:
"""Tests for adjust_points merchant_owner role requirement (Item 3)."""
def test_staff_cannot_adjust_points(
self, client, staff_headers, staff_user_setup
):
"""Staff (non-owner) gets 403 when adjusting points."""
card = staff_user_setup["card"]
response = client.post(
f"{BASE}/cards/{card.id}/points/adjust",
json={"points_delta": 100, "reason": "Staff trying to adjust"},
headers=staff_headers,
)
assert response.status_code == 403
def test_owner_can_adjust_points(
self, client, loyalty_store_headers, loyalty_store_setup
):
"""Merchant owner can adjust points."""
card = loyalty_store_setup["card"]
response = client.post(
f"{BASE}/cards/{card.id}/points/adjust",
json={"points_delta": 50, "reason": "Owner adjustment test"},
headers=loyalty_store_headers,
)
assert response.status_code == 200
data = response.json()
assert data["success"] is True