fix(auth): implement fixed-window rate limiting

- Update TokenAuth to track window start time
- Reset rate limit counter after 1 hour window
- Fixes permanent token blocking
- Adds unit test for rate limit logic

Signed-off-by: Scott R. Shinn <scott@atomicorp.com>

Author: atomicturtle

server/auth.py

@@ -23,7 +23,7 @@ def __init__(self, config_file: str = '/etc/chelon/chelon.conf'):
         """Initialize token auth"""
         self.config_file = Path(config_file)
         self.tokens_file = Path('/var/lib/chelon/tokens.json')
-        self.rate_limits = {}  # token_id -> request count
+        self.rate_limits = {}  # token_id -> {'count': int, 'window_start': float}
 
         # Load tokens
         self.tokens = self._load_tokens()
@@ -122,14 +122,29 @@ def validate_token(self, token: str) -> Dict:
         if secret_hash != token_info['secret_hash']:
             raise ValueError("Invalid token secret")
 
-        # Check rate limit (simple implementation)
-        # TODO: Implement proper time-based rate limiting
-        request_count = self.rate_limits.get(token_id, 0)
-        if request_count >= token_info['rate_limit']:
+        # Check rate limit (Fixed Window)
+        now = datetime.now(UTC)
+        window_size = 3600  # 1 hour in seconds
+        
+        limit_data = self.rate_limits.get(token_id, {
+            'count': 0,
+            'window_start': now.timestamp()
+        })
+        
+        # Check if window has expired
+        if now.timestamp() - limit_data['window_start'] > window_size:
+            # Reset window
+            limit_data = {
+                'count': 0,
+                'window_start': now.timestamp()
+            }
+        
+        if limit_data['count'] >= token_info['rate_limit']:
             raise ValueError("Rate limit exceeded")
 
         # Increment request count
-        self.rate_limits[token_id] = request_count + 1
+        limit_data['count'] += 1
+        self.rate_limits[token_id] = limit_data
 
         # Update last used timestamp
         token_info['last_used'] = datetime.now(UTC).isoformat()

tests/test_rate_limit.py

@@ -0,0 +1,102 @@
+
+import sys
+import os
+import shutil
+import tempfile
+import unittest
+from unittest.mock import patch, MagicMock
+from datetime import datetime, timedelta, timezone
+
+# Add server to path
+sys.path.insert(0, '/home/sshinn/src/chelon/server')
+
+from auth import TokenAuth
+
+class TestRateLimit(unittest.TestCase):
+    def setUp(self):
+        self.temp_dir = tempfile.mkdtemp()
+        self.config_file = os.path.join(self.temp_dir, 'chelon.conf')
+        self.tokens_file = os.path.join(self.temp_dir, 'tokens.json')
+        
+        # Mock paths in TokenAuth
+        self.auth_patcher = patch('auth.Path')
+        self.mock_path = self.auth_patcher.start()
+        
+        # We need to make sure tokens_file path resolves to our temp file
+        # This is scanning for calls to Path('/var/lib/chelon/tokens.json')
+        # Easier way: subclass or just patch the attributes after init if possible,
+        # but init loads tokens.
+        
+        with open(self.config_file, 'w') as f:
+            f.write("TEST=1\n")
+            
+    def tearDown(self):
+        self.auth_patcher.stop()
+        shutil.rmtree(self.temp_dir)
+
+    def test_rate_limit_window(self):
+        # We'll just patch the tokens file attribute and load_tokens method to avoid filesystem issues
+        # effectively testing the logic in validate_token
+        
+        with patch('auth.TokenAuth._load_tokens', return_value={}) as mock_load, \
+             patch('auth.TokenAuth._save_tokens'):
+            
+            auth = TokenAuth(self.config_file)
+            
+            # Manually inject a token
+            token_id = "test-token"
+            secret = "secret"
+            import hashlib
+            secret_hash = hashlib.sha256(secret.encode()).hexdigest()
+            
+            auth.tokens = {
+                token_id: {
+                    'secret_hash': secret_hash,
+                    'permissions': ['sign:rpm'],
+                    'rate_limit': 2,
+                    'created': datetime.now(timezone.utc).isoformat(),
+                    'last_used': None
+                }
+            }
+            
+            token_str = f"{token_id}:{secret}"
+            
+            # Start time: T=0
+            start_time = datetime(2026, 1, 1, 10, 0, 0, tzinfo=timezone.utc)
+            
+            with patch('auth.datetime') as mock_datetime:
+                mock_datetime.now.return_value = start_time
+                
+                # Request 1 (Count: 1) - Should succeed
+                print("Request 1...")
+                auth.validate_token(token_str)
+                
+                # Request 2 (Count: 2) - Should succeed
+                print("Request 2...")
+                auth.validate_token(token_str)
+                
+                # Request 3 (Count: 3 > Limit 2) - Should fail
+                print("Request 3 (Should fail)...")
+                with self.assertRaises(ValueError) as cm:
+                    auth.validate_token(token_str)
+                self.assertEqual(str(cm.exception), "Rate limit exceeded")
+                
+                # Advance time by 30 mins (Still in window)
+                mock_datetime.now.return_value = start_time + timedelta(minutes=30)
+                
+                # Request 4 (Should still fail)
+                print("Request 4 (30m later, Should fail)...")
+                with self.assertRaises(ValueError) as cm:
+                    auth.validate_token(token_str)
+                
+                # Advance time by 61 mins (New window)
+                mock_datetime.now.return_value = start_time + timedelta(minutes=61)
+                
+                # Request 5 (Reset count: 1) - Should succeed
+                print("Request 5 (1h+ later, Should succeed)...")
+                auth.validate_token(token_str)
+                
+                print("Rate limit test passed!")
+
+if __name__ == '__main__':
+    unittest.main()