sync: 181 file da Baida98/AI@80e22d8c (2026-08-25 12:01 UTC) [deploy-all]

api/auth_guard.py

@@ -124,15 +124,20 @@ def _prune_expired_rate_keys(now: float, window_s: float) -> None:
 def _rate_key(role: int, token_header: str | None, client_ip: str | None = None) -> str:
     """Chiave rate limiter: hash(role + discriminante) — non espone token né IP in chiaro.
 
-    GAP-AUTH-FIX: USER (role=0) usa client_ip come discriminante — bucket per IP,
-    non bucket globale condiviso. Previene DoS a costo zero (un client svuota tutti).
-    Ruoli autenticati (MACHINE/OPERATOR/ADMIN) continuano a usare il token hash.
+    USER usa sempre l'IP come discriminante. MACHINE usa l'IP quando il proxy
+    fidato lo inoltra: Cloudflare usa un unico token interno per tutti i browser,
+    quindi il solo token renderebbe globale il limite di 30 richieste/minuto.
+    In assenza di IP attestato, MACHINE conserva il fallback per-token. OPERATOR
+    e ADMIN mantengono il bucket per-token.
     """
     if role == 0:
-        # USER: discrimina per IP — ogni client ha il proprio bucket
+        # USER: bucket per IP, mai globale condiviso.
         raw = f"0:{client_ip or 'unknown'}"
+    elif role == 1 and client_ip:
+        # MACHINE via proxy fidato: separa gli utenti dietro INTERNAL_TOKEN.
+        raw = f"1:{client_ip}"
     else:
-        # Ruoli autenticati: discrimina per token (più preciso dell'IP)
+        # Chiamate server-to-server e ruoli elevati: bucket per token.
         raw = f"{role}:{token_header or 'anonymous'}"
     return _rl_hash.sha256(raw.encode()).hexdigest()[:16]
 

tests/test_security.py

@@ -132,6 +132,32 @@ class TestRequireRole(unittest.TestCase):
                 _run(dep(req_mock, resolved=self.AR.USER))
 
 
+class TestMachineRateLimitClientBuckets(unittest.TestCase):
+    """SEC-3B: il token proxy MACHINE non deve condividere un unico bucket pubblico."""
+
+    def setUp(self):
+        try:
+            from api.auth_guard import _rate_key
+            self.rate_key = _rate_key
+        except ImportError as e:
+            self.skipTest(str(e))
+
+    def test_machine_uses_distinct_attested_client_ips(self):
+        first = self.rate_key(1, "shared-internal-token", "198.51.100.10")
+        second = self.rate_key(1, "shared-internal-token", "198.51.100.11")
+        self.assertNotEqual(first, second)
+
+    def test_machine_without_attested_ip_keeps_token_bucket(self):
+        first = self.rate_key(1, "shared-internal-token", None)
+        second = self.rate_key(1, "shared-internal-token", None)
+        self.assertEqual(first, second)
+
+    def test_operator_remains_token_scoped(self):
+        first = self.rate_key(2, "operator-token", "198.51.100.10")
+        second = self.rate_key(2, "operator-token", "198.51.100.11")
+        self.assertEqual(first, second)
+
+
 class TestBrowserEndpointAuth(unittest.TestCase):
     """SEC-5: endpoint browser stateless non fail-open (GAP-1-fix)."""