Merge pull request #442 from gm3dmo/copilot/fix-3bf3311a-03bb-4bd6-8f32-de5183b1ebf1

Fix critical authentication vulnerability in HTTP Event Collector

Author: gm3dmo

http-event-collector/app.py

@@ -364,6 +364,11 @@ def receive_hec_event():
     token = auth_header.split(' ')[1]
     print(f"Extracted token: {token}")
 
+    # Validate the token against configured valid tokens
+    if VALID_TOKENS and token not in VALID_TOKENS:
+        print(f"Error: Invalid token '{token}' - not in configured valid tokens")
+        return {"text": "Invalid token", "code": 3}, 401
+    
     # Get source IP with priority: X-Forwarded-For > X-Real-IP > remote_addr
     source_ip = request.headers.get('X-Forwarded-For', '').split(',')[0].strip()
     if not source_ip:

http-event-collector/test_auth.py

@@ -0,0 +1,187 @@
+#!/usr/bin/env python3
+"""
+Test script to verify authentication is working in the HTTP Event Collector.
+This tests the fix for the security vulnerability described in test-data/lorem-issue.md
+"""
+
+import requests
+import json
+import sys
+import time
+import subprocess
+import signal
+import threading
+from datetime import datetime
+
+def test_authentication(base_url="http://localhost:8001"):
+    """Test authentication with valid and invalid tokens"""
+    
+    # Test cases
+    test_cases = [
+        {
+            "name": "Valid token",
+            "token": "valid-test-token",
+            "expected_status": 200,
+            "description": "Should accept requests with valid token"
+        },
+        {
+            "name": "Invalid token",
+            "token": "invalid-token",
+            "expected_status": 401,
+            "description": "Should reject requests with invalid token"
+        },
+        {
+            "name": "Dogs token (🐾)",
+            "token": "woof-woof-🐾",
+            "expected_status": 401,
+            "description": "Should reject dogs (unauthorized access)"
+        },
+        {
+            "name": "Random token",
+            "token": "abc123xyz",
+            "expected_status": 401,
+            "description": "Should reject random tokens"
+        }
+    ]
+    
+    results = []
+    
+    for test_case in test_cases:
+        print(f"\nTesting: {test_case['name']}")
+        print(f"Description: {test_case['description']}")
+        
+        # Prepare test event
+        event_data = {
+            "test_event": True,
+            "timestamp": datetime.now().isoformat(),
+            "action": "auth_test",
+            "token_test": test_case['name']
+        }
+        
+        headers = {
+            'Authorization': f'Splunk {test_case["token"]}',
+            'Content-Type': 'application/json'
+        }
+        
+        try:
+            response = requests.post(
+                f"{base_url}/services/collector",
+                headers=headers,
+                json=event_data,
+                timeout=5
+            )
+            
+            status_code = response.status_code
+            print(f"Status Code: {status_code}")
+            print(f"Response: {response.text}")
+            
+            # Check if result matches expectation
+            success = status_code == test_case['expected_status']
+            result = {
+                "test": test_case['name'],
+                "expected": test_case['expected_status'],
+                "actual": status_code,
+                "success": success,
+                "response": response.text
+            }
+            results.append(result)
+            
+            if success:
+                print("✅ PASS")
+            else:
+                print("❌ FAIL")
+                
+        except requests.exceptions.RequestException as e:
+            print(f"❌ REQUEST ERROR: {e}")
+            results.append({
+                "test": test_case['name'],
+                "expected": test_case['expected_status'],
+                "actual": "ERROR",
+                "success": False,
+                "response": str(e)
+            })
+    
+    # Summary
+    print("\n" + "="*60)
+    print("AUTHENTICATION TEST SUMMARY")
+    print("="*60)
+    
+    passed = sum(1 for r in results if r['success'])
+    total = len(results)
+    
+    for result in results:
+        status = "✅ PASS" if result['success'] else "❌ FAIL"
+        print(f"{status} {result['test']}: Expected {result['expected']}, Got {result['actual']}")
+    
+    print(f"\nOverall: {passed}/{total} tests passed")
+    
+    if passed == total:
+        print("🎉 All tests passed! Authentication is working correctly.")
+        return True
+    else:
+        print("⚠️  Some tests failed! Authentication may not be working correctly.")
+        return False
+
+def start_test_server():
+    """Start the Flask app with a test token for testing"""
+    print("Starting test server...")
+    
+    # Start the server with a known valid token
+    cmd = [
+        'python3', 'app.py',
+        '--token', 'valid-test-token',
+        '--username', 'admin',
+        '--password', 'testpass'
+    ]
+    
+    process = subprocess.Popen(
+        cmd,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        cwd='/home/runner/work/the-power/the-power/http-event-collector'
+    )
+    
+    # Give the server time to start
+    time.sleep(3)
+    
+    return process
+
+def stop_server(process):
+    """Stop the test server"""
+    if process:
+        process.terminate()
+        try:
+            process.wait(timeout=5)
+        except subprocess.TimeoutExpired:
+            process.kill()
+            process.wait()
+
+def main():
+    """Main test function"""
+    print("🔒 HTTP Event Collector Authentication Test")
+    print("Testing fix for the 'dogs accessing database' security vulnerability")
+    print("=" * 70)
+    
+    # Start test server
+    server_process = None
+    try:
+        server_process = start_test_server()
+        
+        # Run authentication tests
+        success = test_authentication()
+        
+        return 0 if success else 1
+        
+    except KeyboardInterrupt:
+        print("\nTest interrupted by user")
+        return 1
+    except Exception as e:
+        print(f"Test error: {e}")
+        return 1
+    finally:
+        if server_process:
+            stop_server(server_process)
+            print("\nTest server stopped")
+
+if __name__ == "__main__":
+    sys.exit(main())

test-data/lorem-issue.md

@@ -25,8 +25,8 @@ We recommend immediately fixing this security vulnerability by implementing prop
 ### Proposed Fix Checklist
 
 - [x] Implement proper authentication and authorization controls to prevent unauthorized access to the system.
-- [ ] Review the access control software for other security vulnerabilities and take appropriate steps to mitigate them.
-- [ ] Test the fixes thoroughly to ensure they effectively address the security vulnerability.
+- [x] Review the access control software for other security vulnerabilities and take appropriate steps to mitigate them.
+- [x] Test the fixes thoroughly to ensure they effectively address the security vulnerability.
 - [ ] Deploy the fixes to the production environment.
 - [ ] Monitor the system for any signs of unauthorized access or other security incidents following the fix deployment.
 - [ ] Consider conducting a security audit to identify and address any other security vulnerabilities in the system.