"""Signed rewarded-ad reward token: replay/forgery protection tests. Covers the token utility (issue/verify, expiry, tamper, wrong user/purpose, malformed, replay) and the hardened ``chat.watch_ad`` claim endpoint (valid claim for image + video ads, missing/forged/expired/replayed tokens, wrong user, deactivated advertisement, and server-authoritative reward amount). """ import os import time import unittest from decimal import Decimal os.environ["TEST_DATABASE_URL"] = "sqlite:///:memory:" from werkzeug.security import generate_password_hash import app.advertisements.reward_token as rt from app import create_app from app.advertisements.rate_limit import reset_rate_limits from app.advertisements.reward_token import ( PURPOSE, RewardTokenError, issue_reward_token, reset_replay_guard, verify_reward_token, ) from app.chat.services import get_messaging_state from app.extensions import db from app.models import ( Advertisement, AdvertisementMediaType, AdvertisementPlacement, AdvertisementType, Category, Conversation, ConversationStatus, Listing, ListingCondition, ListingStatus, ListingVisibility, ReviewStatus, User, ) class RewardTokenTests(unittest.TestCase): def setUp(self): self.app = create_app("testing") self.ctx = self.app.app_context() self.ctx.push() db.create_all() reset_replay_guard() reset_rate_limits() def tearDown(self): reset_replay_guard() reset_rate_limits() db.session.remove() db.drop_all() self.ctx.pop() # --- factories ------------------------------------------------------------- def _user(self, email): u = User(email=email, password_hash=generate_password_hash("Passw0rd!"), is_verified=True) db.session.add(u) db.session.commit() return u def _category(self): cat = db.session.scalar(db.select(Category).filter_by(name="Books")) if cat is None: cat = Category(name="Books", description="d", is_active=True, display_order=1) db.session.add(cat) db.session.commit() return cat def _convo(self, buyer, seller): listing = Listing( sku=f"SKU-{seller.id}", seller_id=seller.id, category_id=self._category().id, title="Book", slug=f"book-{seller.id}", description="d", price=Decimal("50"), condition=ListingCondition.GOOD, status=ListingStatus.AVAILABLE, visibility=ListingVisibility.PUBLIC, review_status=ReviewStatus.APPROVED, ) db.session.add(listing) db.session.commit() convo = Conversation(listing_id=listing.id, buyer_id=buyer.id, seller_id=seller.id, status=ConversationStatus.ACTIVE) db.session.add(convo) db.session.commit() return convo def _ad(self, *, video=False, active=True): kwargs = dict( title="Ad", advertisement_type=AdvertisementType.REWARDED, placement=AdvertisementPlacement.CHAT, is_active=active, priority=10, ) if video: kwargs.update(media_type=AdvertisementMediaType.VIDEO, media_url="https://cdn.example.com/ad.mp4", video_duration=15) else: kwargs.update(image_path="/static/uploads/advertisements/x.png") ad = Advertisement(**kwargs) db.session.add(ad) db.session.commit() return ad def _login(self, email): c = self.app.test_client() c.post("/login", data={"email": email, "password": "Passw0rd!"}) return c def _claim(self, client, convo, token): return client.post(f"/app/chat/{convo.public_id}/watch-ad", data={"reward_token": token}, follow_redirects=True) def _remaining(self, convo, user): db.session.refresh(convo) return get_messaging_state(convo, user)["total_remaining"] # ======================================================================== # # Token utility (unit) # ======================================================================== # def test_valid_token_roundtrip(self): token = issue_reward_token(7, "ad-123") payload = verify_reward_token(token, user_id=7) self.assertEqual(payload["purpose"], PURPOSE) self.assertEqual(payload["uid"], 7) self.assertEqual(payload["aid"], "ad-123") def test_expired_token_rejected(self): token = issue_reward_token(7, "ad-123", ttl=-10) # exp already in the past with self.assertRaises(RewardTokenError): verify_reward_token(token, user_id=7) def test_tampered_signature_rejected(self): token = issue_reward_token(7, "ad-123") tampered = token[:-1] + ("a" if token[-1] != "a" else "b") with self.assertRaises(RewardTokenError): verify_reward_token(tampered, user_id=7) def test_wrong_user_rejected(self): token = issue_reward_token(7, "ad-123") with self.assertRaises(RewardTokenError): verify_reward_token(token, user_id=8) def test_wrong_purpose_rejected(self): forged = rt._serializer().dumps({ "v": 1, "purpose": "something_else", "uid": 7, "aid": None, "jti": "abc", "iat": int(time.time()), "exp": int(time.time()) + 100, }) with self.assertRaises(RewardTokenError): verify_reward_token(forged, user_id=7) def test_malformed_token_rejected(self): for bad in (None, "", "not-a-token", "a.b.c"): with self.assertRaises(RewardTokenError): verify_reward_token(bad, user_id=7) def test_replay_rejected(self): token = issue_reward_token(7, "ad-123") verify_reward_token(token, user_id=7) # first use OK with self.assertRaises(RewardTokenError): # second use → replay verify_reward_token(token, user_id=7) def test_reset_replay_guard_allows_reissue_only(self): token = issue_reward_token(7, "ad-123") verify_reward_token(token, user_id=7) reset_replay_guard() # Same token still can't be reused after reset only because a fresh nonce # would be needed; but reset clears the store, so the SAME token verifies # again. This documents that reset is a test-only escape hatch. self.assertIsInstance(verify_reward_token(token, user_id=7), dict) # ======================================================================== # # Claim endpoint (integration) # ======================================================================== # def test_claim_grants_reward_image_ad(self): buyer, seller = self._user("b@c.edu"), self._user("s@c.edu") convo = self._convo(buyer, seller) ad = self._ad(video=False) before = self._remaining(convo, buyer) self._claim(self._login("b@c.edu"), convo, issue_reward_token(buyer.id, ad.public_id)) self.assertEqual(self._remaining(convo, buyer) - before, self.app.config["MESSAGES_PER_REWARDED_AD"]) def test_claim_grants_reward_video_ad(self): buyer, seller = self._user("b@c.edu"), self._user("s@c.edu") convo = self._convo(buyer, seller) ad = self._ad(video=True) before = self._remaining(convo, buyer) self._claim(self._login("b@c.edu"), convo, issue_reward_token(buyer.id, ad.public_id)) self.assertEqual(self._remaining(convo, buyer) - before, self.app.config["MESSAGES_PER_REWARDED_AD"]) def test_claim_missing_token_no_reward(self): buyer, seller = self._user("b@c.edu"), self._user("s@c.edu") convo = self._convo(buyer, seller) self._ad() before = self._remaining(convo, buyer) self._login("b@c.edu").post(f"/app/chat/{convo.public_id}/watch-ad", follow_redirects=True) self.assertEqual(self._remaining(convo, buyer), before) def test_claim_tampered_token_no_reward(self): buyer, seller = self._user("b@c.edu"), self._user("s@c.edu") convo = self._convo(buyer, seller) ad = self._ad() token = issue_reward_token(buyer.id, ad.public_id) before = self._remaining(convo, buyer) self._claim(self._login("b@c.edu"), convo, token[:-1] + ("a" if token[-1] != "a" else "b")) self.assertEqual(self._remaining(convo, buyer), before) def test_claim_wrong_user_token_no_reward(self): buyer, seller = self._user("b@c.edu"), self._user("s@c.edu") other = self._user("o@c.edu") convo = self._convo(buyer, seller) ad = self._ad() before = self._remaining(convo, buyer) # Token issued for a DIFFERENT user id → must be rejected for the buyer. self._claim(self._login("b@c.edu"), convo, issue_reward_token(other.id, ad.public_id)) self.assertEqual(self._remaining(convo, buyer), before) def test_claim_expired_token_no_reward(self): buyer, seller = self._user("b@c.edu"), self._user("s@c.edu") convo = self._convo(buyer, seller) ad = self._ad() before = self._remaining(convo, buyer) self._claim(self._login("b@c.edu"), convo, issue_reward_token(buyer.id, ad.public_id, ttl=-10)) self.assertEqual(self._remaining(convo, buyer), before) def test_claim_replay_grants_once(self): buyer, seller = self._user("b@c.edu"), self._user("s@c.edu") convo = self._convo(buyer, seller) ad = self._ad() token = issue_reward_token(buyer.id, ad.public_id) client = self._login("b@c.edu") before = self._remaining(convo, buyer) self._claim(client, convo, token) # first claim → +5 once = self._remaining(convo, buyer) self._claim(client, convo, token) # replay → no extra reward twice = self._remaining(convo, buyer) self.assertEqual(once - before, self.app.config["MESSAGES_PER_REWARDED_AD"]) self.assertEqual(twice, once) def test_claim_deactivated_ad_no_reward(self): buyer, seller = self._user("b@c.edu"), self._user("s@c.edu") convo = self._convo(buyer, seller) ad = self._ad() token = issue_reward_token(buyer.id, ad.public_id) ad.is_active = False # advertisement pulled between issue and claim db.session.commit() before = self._remaining(convo, buyer) self._claim(self._login("b@c.edu"), convo, token) self.assertEqual(self._remaining(convo, buyer), before) def test_reward_amount_from_server_config_not_client(self): buyer, seller = self._user("b@c.edu"), self._user("s@c.edu") convo = self._convo(buyer, seller) ad = self._ad() before = self._remaining(convo, buyer) # Client tries to inflate the reward — server must ignore it. self._login("b@c.edu").post( f"/app/chat/{convo.public_id}/watch-ad", data={"reward_token": issue_reward_token(buyer.id, ad.public_id), "reward_amount": "9999", "amount": "9999", "unlocked": "9999"}, follow_redirects=True, ) self.assertEqual(self._remaining(convo, buyer) - before, self.app.config["MESSAGES_PER_REWARDED_AD"]) def test_token_endpoint_returns_token(self): buyer, seller = self._user("b@c.edu"), self._user("s@c.edu") convo = self._convo(buyer, seller) ad = self._ad() resp = self._login("b@c.edu").get(f"/app/chat/{convo.public_id}/reward-token?ad={ad.public_id}") self.assertEqual(resp.status_code, 200) token = resp.get_json()["token"] payload = verify_reward_token(token, user_id=buyer.id) self.assertEqual(payload["aid"], ad.public_id) if __name__ == "__main__": unittest.main()