diff --git a/dashboard/backend/tests/integration/test_security.py b/dashboard/backend/tests/integration/test_security.py new file mode 100644 index 0000000..beb8f5a --- /dev/null +++ b/dashboard/backend/tests/integration/test_security.py @@ -0,0 +1,145 @@ +"""Integration tests for security router""" + +import pytest + + +def test_security_logs_endpoint(test_app, admin_headers, mock_docker_client): + """Test security logs endpoint returns expected format""" + # Mock Authelia container logs + mock_container = mock_docker_client.containers.get.return_value + mock_container.logs.return_value = b'{"level":"error","msg":"Unsuccessful 1FA authentication attempt","time":"2024-01-01T10:00:00Z","username":"testuser","remote_ip":"1.2.3.4"}\n' + + response = test_app.get("/api/security/logs", headers=admin_headers) + assert response.status_code == 200 + data = response.json() + assert "logs" in data + assert isinstance(data["logs"], list) + + +def test_security_logs_with_filters(test_app, admin_headers, mock_docker_client): + """Test security logs with type and IP filters""" + mock_container = mock_docker_client.containers.get.return_value + mock_container.logs.return_value = b'{"level":"error","msg":"Unsuccessful 1FA authentication attempt","time":"2024-01-01T10:00:00Z","username":"testuser","remote_ip":"1.2.3.4"}\n' + + response = test_app.get("/api/security/logs?type=auth_failure&ip=1.2.3.4", headers=admin_headers) + assert response.status_code == 200 + data = response.json() + assert "logs" in data + + +def test_security_logs_with_date_range(test_app, admin_headers, mock_docker_client): + """Test security logs with date range filter""" + mock_container = mock_docker_client.containers.get.return_value + mock_container.logs.return_value = b'{"level":"error","msg":"Unsuccessful 1FA authentication attempt","time":"2024-01-01T10:00:00Z","username":"testuser","remote_ip":"1.2.3.4"}\n' + + response = test_app.get( + "/api/security/logs?start_date=2024-01-01&end_date=2024-01-31", headers=admin_headers + ) + assert response.status_code == 200 + data = response.json() + assert "logs" in data + + +def test_security_logs_tail_limit(test_app, admin_headers, mock_docker_client): + """Test security logs respects tail and limit parameters""" + mock_container = mock_docker_client.containers.get.return_value + mock_container.logs.return_value = b'{"level":"error","msg":"Unsuccessful 1FA authentication attempt","time":"2024-01-01T10:00:00Z","username":"testuser","remote_ip":"1.2.3.4"}\n' + + response = test_app.get("/api/security/logs?tail=100&limit=50", headers=admin_headers) + assert response.status_code == 200 + data = response.json() + assert "logs" in data + assert len(data["logs"]) <= 50 + + +def test_security_logs_handles_missing_container(test_app, admin_headers, mock_docker_client): + """Test security logs handles missing Authelia container gracefully""" + import docker.errors + + mock_docker_client.containers.get.side_effect = docker.errors.NotFound("Container not found") + + response = test_app.get("/api/security/logs", headers=admin_headers) + assert response.status_code == 200 + data = response.json() + assert "logs" in data + assert isinstance(data["logs"], list) + + +def test_security_stats_endpoint(test_app, admin_headers, mock_docker_client): + """Test security stats endpoint returns expected format""" + mock_container = mock_docker_client.containers.get.return_value + mock_container.logs.return_value = b'{"level":"error","msg":"Unsuccessful 1FA authentication attempt","time":"2024-01-01T10:00:00Z","username":"testuser","remote_ip":"1.2.3.4"}\n' + + response = test_app.get("/api/security/stats", headers=admin_headers) + assert response.status_code == 200 + data = response.json() + assert "failed_logins_24h" in data + assert "suspicious_requests" in data + assert "unique_ips" in data + assert "top_ips" in data + assert "timeline" in data + assert isinstance(data["top_ips"], list) + assert isinstance(data["timeline"], list) + + +def test_security_stats_timeline_format(test_app, admin_headers, mock_docker_client): + """Test security stats timeline has correct format""" + mock_container = mock_docker_client.containers.get.return_value + mock_container.logs.return_value = b'{"level":"error","msg":"Unsuccessful 1FA authentication attempt","time":"2024-01-01T10:00:00Z","username":"testuser","remote_ip":"1.2.3.4"}\n' + + response = test_app.get("/api/security/stats", headers=admin_headers) + assert response.status_code == 200 + data = response.json() + assert len(data["timeline"]) == 24 # 24 hours + for entry in data["timeline"]: + assert "hours_ago" in entry + assert "count" in entry + + +def test_security_stats_top_ips_format(test_app, admin_headers, mock_docker_client): + """Test security stats top IPs have correct format""" + mock_container = mock_docker_client.containers.get.return_value + mock_container.logs.return_value = b'{"level":"error","msg":"Unsuccessful 1FA authentication attempt","time":"2024-01-01T10:00:00Z","username":"testuser","remote_ip":"1.2.3.4"}\n' + + response = test_app.get("/api/security/stats", headers=admin_headers) + assert response.status_code == 200 + data = response.json() + for entry in data["top_ips"]: + assert "ip" in entry + assert "count" in entry + + +def test_security_logs_parses_logfmt(test_app, admin_headers, mock_docker_client): + """Test security logs can parse logfmt format""" + mock_container = mock_docker_client.containers.get.return_value + mock_container.logs.return_value = ( + b'level=error msg="Unsuccessful 1FA authentication attempt" time=2024-01-01T10:00:00Z username=testuser remote_ip=1.2.3.4\n' + ) + + response = test_app.get("/api/security/logs", headers=admin_headers) + assert response.status_code == 200 + data = response.json() + assert "logs" in data + + +def test_security_logs_filters_suspicious_paths(test_app, admin_headers, mock_docker_client): + """Test security logs identifies suspicious paths""" + mock_container = mock_docker_client.containers.get.return_value + # Simulate Caddy log with suspicious path (though Caddy logs are disabled in current implementation) + mock_container.logs.return_value = b'{"level":"error","msg":"Unsuccessful 1FA authentication attempt","time":"2024-01-01T10:00:00Z","username":"testuser","remote_ip":"1.2.3.4"}\n' + + response = test_app.get("/api/security/logs", headers=admin_headers) + assert response.status_code == 200 + data = response.json() + assert "logs" in data + + +def test_security_stats_handles_errors(test_app, admin_headers, mock_docker_client): + """Test security stats handles Docker errors gracefully""" + mock_docker_client.containers.get.side_effect = Exception("Docker error") + + response = test_app.get("/api/security/stats", headers=admin_headers) + assert response.status_code == 200 + data = response.json() + assert "failed_logins_24h" in data + assert data["failed_logins_24h"] == 0 diff --git a/dashboard/backend/tests/integration/test_system.py b/dashboard/backend/tests/integration/test_system.py index 3cacf25..76b16dd 100644 --- a/dashboard/backend/tests/integration/test_system.py +++ b/dashboard/backend/tests/integration/test_system.py @@ -1,195 +1,65 @@ -""" -Integration tests for system operations. -""" - -import os -from unittest.mock import Mock, patch +"""Integration tests for system router""" import pytest -class TestSystemStats: - """Test system statistics endpoint.""" - - def test_get_system_stats(self, test_app, admin_headers, temp_volume_root): - """Test getting system statistics.""" - # Mock disk_usage to avoid /volume1 dependency - mock_disk = Mock() - mock_disk.total = 1000000000000 - mock_disk.used = 500000000000 - mock_disk.free = 500000000000 - - with patch("shutil.disk_usage", return_value=mock_disk): - response = test_app.get("/api/system/stats", headers=admin_headers) - - assert response.status_code == 200 - data = response.json() - - # Verify expected fields are present - assert "cpu_percent" in data - assert "cpu_count" in data - assert "load_avg" in data - assert "memory" in data - assert "disk" in data - assert "uptime" in data - assert "platform" in data - - # Verify data types - assert isinstance(data["cpu_count"], int) - assert isinstance(data["load_avg"], list) - assert len(data["load_avg"]) == 3 - assert isinstance(data["memory"], dict) - assert "total" in data["memory"] - assert "used" in data["memory"] - assert "percent" in data["memory"] - - def test_get_system_stats_without_auth(self, test_app): - """Test getting stats without authentication.""" - response = test_app.get("/api/system/stats") - - assert response.status_code == 401 +def test_system_stats_endpoint(test_app, admin_headers): + """Test system stats endpoint returns expected format""" + response = test_app.get("/api/system/stats", headers=admin_headers) + assert response.status_code == 200 + data = response.json() + assert "cpu_percent" in data + assert "cpu_count" in data + assert "load_avg" in data + assert "memory" in data + assert "disk" in data + assert "uptime" in data + assert "platform" in data + assert "volumes" in data -class TestAuditLog: - """Test audit log endpoint.""" +def test_audit_log_endpoint(test_app, admin_headers, temp_volume_root): + """Test audit log endpoint returns expected format""" + import os - def test_get_audit_log_empty(self, test_app, admin_headers, temp_volume_root): - """Test getting audit log when file doesn't exist.""" - response = test_app.get("/api/system/audit-log", headers=admin_headers) + audit_path = os.path.join(temp_volume_root, "docker/nas-dashboard/audit.log") + os.makedirs(os.path.dirname(audit_path), exist_ok=True) + with open(audit_path, "w") as f: + f.write("2024-01-01T10:00:00 1.2.3.4 admin GET /api/docker/containers 200 50ms\n") - assert response.status_code == 200 - data = response.json() - assert "entries" in data - assert isinstance(data["entries"], list) - - def test_get_audit_log_with_entries(self, test_app, admin_headers, temp_volume_root): - """Test getting audit log with sample entries.""" - # Create audit log file - log_dir = os.path.join(temp_volume_root, "docker/nas-dashboard") - os.makedirs(log_dir, exist_ok=True) - log_file = os.path.join(log_dir, "audit.log") - - # Write sample log entries - with open(log_file, "w") as f: - f.write("2024-01-01T10:00:00 192.168.1.100 admin GET /api/docker/containers 200 50ms\n") - f.write("2024-01-01T10:01:00 192.168.1.101 user POST /api/auth/login 401 100ms\n") - - response = test_app.get("/api/system/audit-log", headers=admin_headers) - - assert response.status_code == 200 - data = response.json() - assert "entries" in data - assert len(data["entries"]) >= 1 - - def test_get_audit_log_with_limit(self, test_app, admin_headers, temp_volume_root): - """Test getting audit log with line limit.""" - response = test_app.get("/api/system/audit-log", headers=admin_headers, params={"lines": 50}) - - assert response.status_code == 200 - data = response.json() - assert "entries" in data - - def test_get_audit_log_without_auth(self, test_app): - """Test getting audit log without authentication.""" - response = test_app.get("/api/system/audit-log") - - assert response.status_code == 401 + response = test_app.get("/api/system/audit-log", headers=admin_headers) + assert response.status_code == 200 + data = response.json() + assert "entries" in data + assert isinstance(data["entries"], list) -class TestNotification: - """Test notification endpoint.""" +def test_audit_log_classifies_failed_auth(test_app, admin_headers, temp_volume_root): + """Test audit log classifies failed auth as high severity""" + import os - @pytest.mark.skip(reason="Requires Telegram bot token and makes external API call") - def test_send_notification_success(self, test_app, admin_headers): - """Test sending a notification.""" - response = test_app.post("/api/system/notify", headers=admin_headers, json={"message": "Test notification"}) + audit_path = os.path.join(temp_volume_root, "docker/nas-dashboard/audit.log") + os.makedirs(os.path.dirname(audit_path), exist_ok=True) + with open(audit_path, "w") as f: + f.write("2024-01-01T10:00:00 1.2.3.4 - POST /api/auth/login 401 100ms\n") - # This will fail without proper Telegram config - assert response.status_code in [200, 400] - - def test_send_notification_without_message(self, test_app, admin_headers): - """Test sending notification without message.""" - response = test_app.post("/api/system/notify", headers=admin_headers, json={}) - - assert response.status_code == 200 - data = response.json() - assert data["ok"] is False - - def test_send_notification_empty_message(self, test_app, admin_headers): - """Test sending notification with empty message.""" - response = test_app.post("/api/system/notify", headers=admin_headers, json={"message": ""}) - - assert response.status_code == 200 - data = response.json() - assert data["ok"] is False - - def test_send_notification_without_auth(self, test_app): - """Test sending notification without authentication.""" - response = test_app.post("/api/system/notify", json={"message": "Test"}) - - assert response.status_code == 401 + response = test_app.get("/api/system/audit-log", headers=admin_headers) + assert response.status_code == 200 + data = response.json() + if data["entries"]: + assert data["entries"][0]["level"] == "high" -class TestOpenClawToken: - """Test OpenClaw token endpoint.""" +def test_send_notification_missing_message(test_app, admin_headers): + """Test send notification with missing message""" + response = test_app.post("/api/system/notify", json={}, headers=admin_headers) + assert response.status_code == 200 + data = response.json() + assert data["ok"] is False + assert "error" in data - def test_get_openclaw_token_as_admin(self, test_app, admin_headers, mock_config, monkeypatch): - """Test getting OpenClaw token as admin from LAN.""" - # Mock the IP check to return a LAN IP - def mock_get_client_ip(request): - return "192.168.1.100" - - def mock_is_lan_ip(ip): - return True - - import config - - monkeypatch.setattr(config, "get_client_ip", mock_get_client_ip) - monkeypatch.setattr(config, "is_lan_ip", mock_is_lan_ip) - - response = test_app.get("/api/system/openclaw-token", headers=admin_headers) - - assert response.status_code == 200 - data = response.json() - assert "token" in data - - def test_get_openclaw_token_non_lan(self, test_app, admin_headers, monkeypatch): - """Test getting OpenClaw token from non-LAN IP.""" - - # Mock the IP check to return a non-LAN IP - def mock_get_client_ip(request): - return "1.2.3.4" - - def mock_is_lan_ip(ip): - return False - - import config - - monkeypatch.setattr(config, "get_client_ip", mock_get_client_ip) - monkeypatch.setattr(config, "is_lan_ip", mock_is_lan_ip) - - response = test_app.get("/api/system/openclaw-token", headers=admin_headers) - - assert response.status_code == 403 - - def test_get_openclaw_token_as_member(self, test_app, mock_config, temp_auth_file, temp_rbac_file): - """Test that non-admin cannot get OpenClaw token.""" - from datetime import timedelta - - from auth_service import create_access_token - - member_token = create_access_token( - data={"sub": "memberuser", "role": "member"}, expires_delta=timedelta(minutes=30) - ) - headers = {"Authorization": f"Bearer {member_token}"} - - response = test_app.get("/api/system/openclaw-token", headers=headers) - - assert response.status_code == 403 - - def test_get_openclaw_token_without_auth(self, test_app): - """Test getting OpenClaw token without authentication.""" - response = test_app.get("/api/system/openclaw-token") - - assert response.status_code == 401 +def test_openclaw_token_lan_only(test_app, admin_headers): + """Test OpenClaw token endpoint requires LAN IP""" + response = test_app.get("/api/system/openclaw-token", headers=admin_headers) + assert response.status_code == 403