detecting-api-enumeration-attacks

name	detecting-api-enumeration-attacks
description	通过监控顺序标识符访问模式和授权失败，检测和防止API枚举攻击，包括BOLA（越权对象访问）和IDOR（不安全的直接对象引用）利用。
domain	cybersecurity
subdomain	api-security
tags	["api-security","enumeration","bola","idor","broken-object-level-authorization","owasp-api-top-10","access-control","rate-limiting"]
version	1.0
author	mahipal
license	Apache-2.0

检测API枚举攻击

概述

API枚举攻击（API Enumeration Attack）是指攻击者通过系统地探测具有顺序或可预测标识符的API端点，来发现和访问未授权资源。越权对象访问（Broken Object Level Authorization，BOLA）在OWASP API安全Top 10中被列为API1:2023，是最严重的API漏洞。攻击者操纵API请求中的对象标识符（用户ID、订单号、账户引用）来绕过授权并访问其他用户的数据。检测需要监控快速顺序访问尝试、授权失败和异常API使用行为的模式。

前置条件

• 已启用日志记录的API网关或反向代理（Kong、AWS API Gateway、Apigee）
• SIEM平台（Splunk、Elastic SIEM或Microsoft Sentinel）
• 可访问包含请求详情的API服务器日志
• 具备API保护能力的Web应用防火墙（WAF）
• 了解API的授权模型和对象标识符方案

检测的攻击模式

1. 顺序ID枚举

攻击者遍历数字或可预测标识符：

GET /api/v1/users/1001 -> 200 OK
GET /api/v1/users/1002 -> 200 OK
GET /api/v1/users/1003 -> 403 Forbidden
GET /api/v1/users/1004 -> 200 OK
GET /api/v1/users/1005 -> 200 OK
...

检测指标：

• 对同一端点进行快速顺序请求，ID递增
• 同一来源混合出现200/403/401响应
• 请求速率超过正常用户行为
• 访问超出已认证用户范围的资源

2. UUID/GUID枚举

即使是非顺序标识符，如果通过其他端点泄露也可被枚举：

# 攻击者首先从列表端点收集UUID
GET /api/v1/posts?page=1  -> 返回包含作者UUID的帖子对象

# 然后使用这些UUID访问受限用户数据
GET /api/v1/users/a3f2c1e4-... -> 私人用户档案
GET /api/v1/users/b7d9e8f1-... -> 私人用户档案

3. 参数篡改枚举

# 以user_id=100认证，尝试访问其他用户的订单
GET /api/v1/orders?user_id=101
GET /api/v1/orders?user_id=102
GET /api/v1/orders?user_id=103

检测规则

Splunk检测查询

# 检测API端点上的顺序ID枚举
index=api_logs sourcetype=api_access
| rex field=uri_path "(?<endpoint>/api/v\d+/\w+/)(?<object_id>\d+)"
| stats count as request_count,
        dc(object_id) as unique_ids,
        values(status_code) as status_codes,
        min(_time) as first_seen,
        max(_time) as last_seen
  by src_ip, endpoint, user_session
| eval time_span = last_seen - first_seen
| eval requests_per_second = request_count / max(time_span, 1)
| where unique_ids > 20 AND requests_per_second > 2
| eval severity = case(
    unique_ids > 100, "critical",
    unique_ids > 50, "high",
    unique_ids > 20, "medium",
    1==1, "low"
  )
| sort - unique_ids
| table src_ip, endpoint, unique_ids, request_count, requests_per_second,
        status_codes, severity

# 通过授权失败模式检测BOLA
index=api_logs sourcetype=api_access status_code IN (401, 403)
| bin _time span=5m
| stats count as failure_count,
        dc(uri_path) as unique_paths,
        values(uri_path) as attempted_paths
  by _time, src_ip, user_id
| where failure_count > 10
| eval attack_type = if(unique_paths > 5, "enumeration", "brute_force")

Elastic SIEM检测规则

{
  "rule": {
    "name": "API Object Enumeration Detection",
    "description": "Detects rapid sequential access to API objects with mixed authorization results",
    "type": "threshold",
    "index": ["api-access-*"],
    "query": {
      "bool": {
        "must": [
          { "regexp": { "url.path": "/api/v[0-9]+/[a-z]+/[0-9]+" } }
        ],
        "should": [
          { "term": { "http.response.status_code": 200 } },
          {      
                
        
      
    
     
       
       
       
             
      
    
        
     
     
       
  

自定义检测脚本

#!/usr/bin/env python3
"""API枚举攻击检测器

分析API访问日志以检测枚举模式，
包括BOLA、IDOR和顺序ID探测。
"""

import re
import sys
import json
from collections import defaultdict
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import List, Dict, Optional

@dataclass
class AccessRecord:
    timestamp: datetime
    source_ip: str
    user_id: Optional[str]
    method: str
    path: str
    status_code: int
    object_id: Optional[str] = None

@dataclass
class EnumerationAlert:
    source_ip: str
    user_id: Optional[str]
    endpoint_pattern: str
    unique_object_ids: int
    total_requests: int
    time_window_seconds: float
    requests_per_second: float
    auth_failure_ratio: float
    severity: str
    attack_type: str
    sample_ids: List[str] = field(default_factory=list)

class EnumerationDetector:
    # 从API路径中提取对象ID的正则表达式模式
    ID_PATTERNS = [
        re.(),           
        re.(), 
        re.(), 
    ]

     ():
        .time_window = timedelta(minutes=time_window_minutes)
        .min_unique_ids = min_unique_ids
        .max_rps = max_requests_per_second
        .access_log: [AccessRecord] = []

     () -> [AccessRecord]:
        
        log_pattern = re.(
            
            
            
            
        )
         = log_pattern.(line)
          :
             

        path = .group()
        object_id = 
         pattern  .ID_PATTERNS:
            id_match = pattern.search(path)
             id_match:
                object_id = id_match.group()
                

         AccessRecord(
            timestamp=datetime.strptime(.group(), ),
            source_ip=.group(),
            user_id=.group()  .group() !=   ,
            method=.group(),
            path=path,
            status_code=(.group()),
            object_id=object_id
        )

     () -> [EnumerationAlert]:
        
        alerts = []

        
        grouped = defaultdict()
         record  records:
             record.object_id:
                
                endpoint = re.sub(, ,
                         re.sub(, , record.path))
                key = (record.source_ip, record.user_id, endpoint)
                grouped[key].append(record)

         (src_ip, user_id, endpoint), records_group  grouped.items():
             (records_group) < .min_unique_ids:
                

            
            records_group.sort(key= r: r.timestamp)

            
            window_start = 
             window_start  ((records_group)):
                window_records = []
                 r  records_group[window_start:]:
                     r.timestamp - records_group[window_start].timestamp <= .time_window:
                        window_records.append(r)

                unique_ids = (r.object_id  r  window_records)
                 (unique_ids) < .min_unique_ids:
                    

                time_span = (window_records[-].timestamp -
                           window_records[].timestamp).total_seconds()
                rps = (window_records) / (time_span, )

                auth_failures = (  r  window_records
                                    r.status_code  (, ))
                failure_ratio = auth_failures / (window_records)

                
                 (unique_ids) > :
                    severity = 
                 (unique_ids) >   failure_ratio > :
                    severity = 
                 (unique_ids) > :
                    severity = 
                :
                    severity = 

                
                ids_list = ([r.object_id  r  window_records
                                   r.object_id  r.object_id.isdigit()])
                is_sequential = ._check_sequential(ids_list)
                attack_type =   is_sequential  

                alert = EnumerationAlert(
                    source_ip=src_ip,
                    user_id=user_id,
                    endpoint_pattern=endpoint,
                    unique_object_ids=(unique_ids),
                    total_requests=(window_records),
                    time_window_seconds=time_span,
                    requests_per_second=(rps, ),
                    auth_failure_ratio=(failure_ratio, ),
                    severity=severity,
                    attack_type=attack_type,
                    sample_ids=(unique_ids)[:]
                )
                alerts.append(alert)
                  

         alerts

     () -> :
        
         (ids) < :
             
        :
            numeric_ids = ((i)  i  ids)
            sequential_count = (
                  i  (, (numeric_ids))
                 numeric_ids[i] - numeric_ids[i-] <= 
            )
             sequential_count / (numeric_ids) > 
         ValueError:
             


 ():
    detector = EnumerationDetector(
        time_window_minutes=,
        min_unique_ids=
    )

    log_file = sys.argv[]  (sys.argv) >   
    records = []
     (log_file, )  f:
         line  f:
            record = detector.parse_log_line(line.strip())
             record:
                records.append(record)

    alerts = detector.analyze(records)

     alerts:
        ()
         alert  alerts:
            ()
            ()
            ()
            ()
            ()
            ()
            ()
            ()
            ()
            ()
    :
        ()


 __name__ == :
    main()

防护控制

服务端授权强制执行

# 始终在数据层验证对象所有权
def get_user_order(request, order_id):
    order = Order.objects.get(id=order_id)
    if order.user_id != request.user.id:
        raise PermissionDenied("无权访问此订单")
    return order

使用不可预测的标识符

import uuid

# 使用UUID替代顺序整数
class Order(Model):
    id = UUIDField(default=uuid.uuid4, primary_key=True)

对每个端点实施速率限制

# Kong每个API路由的速率限制
plugins:
  - name: rate-limiting
    config:
      minute: 30
      policy: redis
      limit_by: credential

参考资料

• OWASP API1:2023 Broken Object Level Authorization: https://owasp.org/API-Security/editions/2023/en/0xa1-broken-object-level-authorization/
• Traceable.ai BOLA Deep Dive: https://www.traceable.ai/blog-post/a-deep-dive-on-the-most-critical-api-vulnerability----bola-broken-object-level-authorization
• Cequence BOLA Prevention: https://www.cequence.ai/solutions/bola-and-enumeration-attack-prevention/
• Cloudflare API Shield BOLA Detection: https://community.cloudflare.com/t/api-shield-new-bola-vulnerability-detection-for-api-shield/883021
• Sycope IDOR Detection via HTTP Traffic Analysis: https://www.sycope.com/post/idor-vulnerability-how-to-detect-an-attack-on-web-applications-through-http-traffic-analysis