(scope, limit=20, window=60)
| 2156 | |
| 2157 | |
| 2158 | def rate_limited(scope, limit=20, window=60): |
| 2159 | now = time.time() |
| 2160 | key = f"{scope}:{client_ip()}" |
| 2161 | with RATE_LIMIT_LOCK: |
| 2162 | bucket = [ts for ts in RATE_LIMIT_BUCKETS.get(key, []) if now - ts < window] |
| 2163 | if len(bucket) >= limit: |
| 2164 | RATE_LIMIT_BUCKETS[key] = bucket |
| 2165 | return True |
| 2166 | bucket.append(now) |
| 2167 | RATE_LIMIT_BUCKETS[key] = bucket |
| 2168 | return False |
| 2169 | |
| 2170 | |
| 2171 | def require_sensitive_request(scope, limit=20, window=60): |
no test coverage detected