Update app.py

app.py

@@ -190,8 +190,10 @@ def _flush_request_log():
     except Exception as e:
         print(f"Warning: Failed to flush request log: {e}")
 
-def _get_recent_requests_from_log(ip_address: str) -> list:
-    """Get recent requests for an IP from the log file (with caching)"""
+def _get_recent_requests_from_log(ip_address: str) -> tuple:
+    """Get recent requests for an IP from the log file (with caching)
+    Returns: (list of ages in seconds, set of user_agents, set of order_numbers)
+    """
     now_time = time.time()
     now_dt = datetime.now(timezone.utc)
 
@@ -210,11 +212,14 @@ def _get_recent_requests_from_log(ip_address: str) -> list:
             _request_log_cache["data"] = log_data
             _request_log_cache["ts"] = now_time
         except Exception:
-            # If log doesn't exist or there's an error, return empty list
-            return []
+            # If log doesn't exist or there's an error, return empty data
+            return [], set(), set()
 
     # Filter for this IP and recent requests
     recent_timestamps = []
+    user_agents = set()
+    order_numbers = set()
+
     for row in log_data:
         if row.get('ip_address') == ip_address:
             try:
@@ -224,18 +229,22 @@ def _get_recent_requests_from_log(ip_address: str) -> list:
                 # Only consider requests from the last hour
                 if age_seconds < 3600:
                     recent_timestamps.append(age_seconds)
+                    if row.get('user_agent'):
+                        user_agents.add(row['user_agent'])
+                    if row.get('order_number'):
+                        order_numbers.add(row['order_number'])
             except (ValueError, KeyError):
                 continue
 
-    return recent_timestamps
+    return recent_timestamps, user_agents, order_numbers
 
-def _check_suspicious_activity(ip_address: str, order_number: str) -> list:
+def _check_suspicious_activity(ip_address: str, order_number: str, user_agent: str = "unknown") -> list:
     """Check for suspicious patterns and return list of flags"""
     flags = []
     now_dt = datetime.now(timezone.utc)
 
     # Get recent requests from the persistent log
-    recent_request_ages = _get_recent_requests_from_log(ip_address)
+    recent_request_ages, user_agents_from_ip, order_numbers_from_ip = _get_recent_requests_from_log(ip_address)
 
     # Also check the buffer for requests not yet written to the log
     for entry in _request_log_buffer:
@@ -245,6 +254,9 @@ def _check_suspicious_activity(ip_address: str, order_number: str) -> list:
                 age_seconds = (now_dt - timestamp).total_seconds()
                 if age_seconds < 3600:
                     recent_request_ages.append(age_seconds)
+                    user_agents_from_ip.add(entry.get('user_agent', 'unknown'))
+                    if entry.get('order_number'):
+                        order_numbers_from_ip.add(entry['order_number'])
             except (ValueError, KeyError):
                 continue
 
@@ -255,16 +267,35 @@ def _check_suspicious_activity(ip_address: str, order_number: str) -> list:
     if total_requests > 20:
         flags.append("HIGH_FREQUENCY")
 
-    # Flag: More than 5 requests in the last minute
+    # Flag: More than 3 requests in the last minute
     last_minute = sum(1 for age in recent_request_ages if age < 60) + 1
-    if last_minute > 5:
+    if last_minute > 3:
         flags.append("RAPID_REQUESTS")
 
-    # Flag: More than 10 requests in the last 5 minutes
+    # Flag: More than 5 requests in the last 5 minutes
     last_5_min = sum(1 for age in recent_request_ages if age < 300) + 1
-    if last_5_min > 10:
+    if last_5_min > 5:
         flags.append("BURST_PATTERN")
 
+    # Flag: Suspicious user agent patterns (low false positive risk)
+    ua_lower = user_agent.lower()
+    suspicious_ua_patterns = [
+        'bot', 'crawler', 'spider', 'scraper', 'curl', 'wget',
+        'python-requests', 'python-urllib', 'go-http-client',
+        'java/', 'okhttp', 'axios', 'node-fetch'
+    ]
+    if any(pattern in ua_lower for pattern in suspicious_ua_patterns):
+        flags.append("SUSPICIOUS_USER_AGENT")
+
+    # Flag: Multiple different user agents from same IP (enumeration attempt)
+    # Only flag if we have enough data to be confident
+    if len(user_agents_from_ip) >= 3 and user_agent not in user_agents_from_ip:
+        flags.append("MULTIPLE_USER_AGENTS")
+
+    # Flag: Trying many different order numbers from same IP
+    if order_number and len(order_numbers_from_ip) >= 5 and order_number not in order_numbers_from_ip:
+        flags.append("ORDER_ENUMERATION")
+
     return flags
 
 # -------------------------
@@ -289,7 +320,16 @@ def claim_c_key(
     order_number = order_number.strip()
 
     # Check for suspicious activity patterns
-    suspicious_flags = _check_suspicious_activity(ip_address, order_number)
+    suspicious_flags = _check_suspicious_activity(ip_address, order_number, user_agent)
+
+    # Apply exponential delay for suspicious requests (anti-abuse measure)
+    # This significantly slows down automated attacks while having minimal impact on legitimate users
+    # Delay formula: 2^(number_of_flags) seconds, capped at 30 seconds
+    # Examples: 1 flag = 2s, 2 flags = 4s, 3 flags = 8s, 4 flags = 16s, 5+ flags = 30s
+    if suspicious_flags:
+        delay_seconds = 2 ** len(suspicious_flags)
+        delay_seconds = min(delay_seconds, 30)  # Cap at 30 seconds
+        time.sleep(delay_seconds)
 
     if not order_number:
         _log_request(order_number, ip_address, user_agent, False, "Empty order number", suspicious_flags)
@@ -417,3 +457,4 @@ with gr.Blocks(title="API") as demo:
 demo.queue()
 demo.launch()
 
+