wehub-resource-sync
156 lines
4.9 KiB
Python
156 lines
4.9 KiB
Python
"""Tests for dashboard token authentication."""
|
|
|
|
import sys
|
|
|
|
import pytest
|
|
import requests
|
|
|
|
|
|
def test_dashboard_request_requires_auth_with_valid_token(
|
|
setup_cluster_with_token_auth,
|
|
):
|
|
"""Test that requests succeed with valid token when auth is enabled."""
|
|
|
|
cluster_info = setup_cluster_with_token_auth
|
|
headers = {"Authorization": f"Bearer {cluster_info['token']}"}
|
|
|
|
response = requests.get(
|
|
f"{cluster_info['dashboard_url']}/api/component_activities",
|
|
headers=headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_dashboard_request_requires_auth_missing_token(setup_cluster_with_token_auth):
|
|
"""Test that requests fail without token when auth is enabled."""
|
|
|
|
cluster_info = setup_cluster_with_token_auth
|
|
|
|
response = requests.get(
|
|
f"{cluster_info['dashboard_url']}/api/component_activities",
|
|
json={"test": "data"},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_dashboard_request_requires_auth_invalid_token(setup_cluster_with_token_auth):
|
|
"""Test that requests fail with invalid token when auth is enabled."""
|
|
|
|
cluster_info = setup_cluster_with_token_auth
|
|
headers = {"Authorization": "Bearer wrong_token_00000000000000000000000000000000"}
|
|
|
|
response = requests.get(
|
|
f"{cluster_info['dashboard_url']}/api/component_activities",
|
|
json={"test": "data"},
|
|
headers=headers,
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_dashboard_request_with_ray_auth_header(setup_cluster_with_token_auth):
|
|
"""Test that requests succeed with valid token in X-Ray-Authorization header."""
|
|
|
|
cluster_info = setup_cluster_with_token_auth
|
|
headers = {"X-Ray-Authorization": f"Bearer {cluster_info['token']}"}
|
|
|
|
response = requests.get(
|
|
f"{cluster_info['dashboard_url']}/api/component_activities",
|
|
headers=headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_authorization_header_takes_precedence(setup_cluster_with_token_auth):
|
|
"""Test that standard Authorization header takes precedence over X-Ray-Authorization."""
|
|
|
|
cluster_info = setup_cluster_with_token_auth
|
|
|
|
# Provide both headers: valid token in Authorization, invalid in X-Ray-Authorization
|
|
headers = {
|
|
"Authorization": f"Bearer {cluster_info['token']}",
|
|
"X-Ray-Authorization": "Bearer invalid_token_000000000000000000000000",
|
|
}
|
|
|
|
# Should succeed because Authorization header takes precedence
|
|
response = requests.get(
|
|
f"{cluster_info['dashboard_url']}/api/component_activities",
|
|
headers=headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
# Now test with invalid Authorization but valid X-Ray-Authorization
|
|
headers = {
|
|
"Authorization": "Bearer invalid_token_000000000000000000000000",
|
|
"X-Ray-Authorization": f"Bearer {cluster_info['token']}",
|
|
}
|
|
|
|
# Should fail because Authorization header takes precedence (even though it's invalid)
|
|
response = requests.get(
|
|
f"{cluster_info['dashboard_url']}/api/component_activities",
|
|
headers=headers,
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_dashboard_auth_disabled(setup_cluster_without_token_auth):
|
|
"""Test that auth is not enforced when AUTH_MODE is disabled."""
|
|
|
|
cluster_info = setup_cluster_without_token_auth
|
|
|
|
response = requests.get(
|
|
f"{cluster_info['dashboard_url']}/api/component_activities",
|
|
json={"test": "data"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_authentication_mode_endpoint_with_token_auth(setup_cluster_with_token_auth):
|
|
"""Test authentication_mode endpoint returns 'token' when auth is enabled."""
|
|
|
|
cluster_info = setup_cluster_with_token_auth
|
|
|
|
# This endpoint should be accessible WITHOUT authentication
|
|
response = requests.get(f"{cluster_info['dashboard_url']}/api/authentication_mode")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"authentication_mode": "token"}
|
|
|
|
|
|
def test_authentication_mode_endpoint_without_auth(setup_cluster_without_token_auth):
|
|
"""Test authentication_mode endpoint returns 'disabled' when auth is off."""
|
|
|
|
cluster_info = setup_cluster_without_token_auth
|
|
|
|
response = requests.get(f"{cluster_info['dashboard_url']}/api/authentication_mode")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"authentication_mode": "disabled"}
|
|
|
|
|
|
def test_authentication_mode_endpoint_is_public(setup_cluster_with_token_auth):
|
|
"""Test authentication_mode endpoint works without Authorization header."""
|
|
|
|
cluster_info = setup_cluster_with_token_auth
|
|
|
|
# Call WITHOUT any authorization header - should still succeed
|
|
response = requests.get(
|
|
f"{cluster_info['dashboard_url']}/api/authentication_mode",
|
|
headers={}, # Explicitly no auth
|
|
)
|
|
|
|
# Should succeed even with token auth enabled
|
|
assert response.status_code == 200
|
|
assert response.json() == {"authentication_mode": "token"}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
sys.exit(pytest.main(["-vv", __file__]))
|