356 lines
15 KiB
Python
356 lines
15 KiB
Python
# tests/test_auth.py — Tests for authentication and authorization
|
|
|
|
import pytest
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Auth status endpoint (public)
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
class TestAuthStatus:
|
|
def test_auth_disabled_in_test_env(self, client):
|
|
"""Auth should be disabled in the test environment."""
|
|
resp = client.get("/api/auth/status")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["auth_enabled"] is False
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Password hashing
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
class TestPasswordHashing:
|
|
def test_hash_and_verify(self):
|
|
from backend.auth.password import hash_password, verify_password
|
|
|
|
password = "SuperSecret123!"
|
|
hashed = hash_password(password)
|
|
assert hashed != password
|
|
assert verify_password(password, hashed) is True
|
|
assert verify_password("WrongPassword", hashed) is False
|
|
|
|
def test_different_salts(self):
|
|
from backend.auth.password import hash_password
|
|
|
|
hash1 = hash_password("test")
|
|
hash2 = hash_password("test")
|
|
# Same password, different salts → different hashes
|
|
assert hash1 != hash2
|
|
|
|
def test_reject_short_passwords(self):
|
|
from backend.auth.password import hash_password
|
|
# Argon2 should handle any length, but we test it doesn't crash
|
|
result = hash_password("ab")
|
|
assert result is not None
|
|
|
|
def test_argon2_memory_recalibrated(self):
|
|
"""BUG-038: memory cost must stay at the OWASP 19 MiB recommendation."""
|
|
from backend.auth.password import (
|
|
ARGON2_MEMORY_COST_KIB,
|
|
ARGON2_PARALLELISM,
|
|
ARGON2_TIME_COST,
|
|
ph,
|
|
)
|
|
|
|
assert ARGON2_MEMORY_COST_KIB == 19456
|
|
assert ARGON2_TIME_COST == 2
|
|
assert ARGON2_PARALLELISM == 1
|
|
assert ph.memory_cost == 19456
|
|
assert ph.time_cost == 2
|
|
assert ph.parallelism == 1
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# JWT Handler
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
class TestJWTHandler:
|
|
def test_create_and_decode_access_token(self):
|
|
from backend.auth.jwt_handler import (
|
|
create_access_token,
|
|
decode_token,
|
|
)
|
|
user = {"username": "testuser", "role": "admin", "vaults": ["*"], "display_name": "Test User"}
|
|
token = create_access_token(user)
|
|
assert token is not None
|
|
payload = decode_token(token)
|
|
assert payload is not None
|
|
assert payload["sub"] == "testuser"
|
|
assert payload["type"] == "access"
|
|
|
|
def test_expired_token(self):
|
|
from backend.auth.jwt_handler import (
|
|
create_access_token,
|
|
decode_token,
|
|
)
|
|
|
|
# Override TTL to a value in the past
|
|
user = {"username": "testuser", "role": "user", "vaults": [], "display_name": "X"}
|
|
# Create token that will be decoded with leeway by PyJWT
|
|
token = create_access_token(user)
|
|
payload = decode_token(token)
|
|
assert payload is not None # Valid when just created
|
|
|
|
def test_refresh_token(self):
|
|
from backend.auth.jwt_handler import (
|
|
create_refresh_token,
|
|
decode_token,
|
|
)
|
|
token, jti = create_refresh_token("testuser")
|
|
assert token is not None
|
|
assert jti is not None
|
|
payload = decode_token(token)
|
|
assert payload["type"] == "refresh"
|
|
|
|
def test_token_revocation(self):
|
|
from backend.auth.jwt_handler import (
|
|
create_refresh_token,
|
|
revoke_token,
|
|
is_token_revoked,
|
|
)
|
|
token, jti = create_refresh_token("testuser")
|
|
assert is_token_revoked(jti) is False
|
|
revoke_token(jti)
|
|
assert is_token_revoked(jti) is True
|
|
|
|
def test_invalid_token(self):
|
|
from backend.auth.jwt_handler import decode_token
|
|
payload = decode_token("not.a.valid.token")
|
|
assert payload is None
|
|
|
|
def test_wrong_type_token(self):
|
|
"""decode_token rejects refresh tokens (type != 'access')."""
|
|
from backend.auth.jwt_handler import (
|
|
create_refresh_token,
|
|
decode_token,
|
|
)
|
|
refresh_token, _ = create_refresh_token("testuser")
|
|
payload = decode_token(refresh_token)
|
|
# decode_token rejects non-access types in middleware path;
|
|
# the raw decode still works but is filtered by require_auth.
|
|
# Here we test that the payload has type='refresh'
|
|
if payload is not None:
|
|
assert payload["type"] == "refresh"
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# User Store
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
class TestUserStore:
|
|
@pytest.fixture(autouse=True)
|
|
def _setup_users(self, tmp_path, monkeypatch):
|
|
"""Use a temp file for users.json."""
|
|
from backend.auth import user_store
|
|
users_file = tmp_path / "users.json"
|
|
monkeypatch.setattr(user_store, "USERS_FILE", users_file)
|
|
# Clear the module-level cache
|
|
user_store._users_cache = None
|
|
yield
|
|
user_store._users_cache = None
|
|
|
|
def test_has_users_false_initially(self):
|
|
from backend.auth.user_store import has_users
|
|
assert has_users() is False
|
|
|
|
def test_create_and_get_user(self):
|
|
from backend.auth.user_store import create_user, get_user, has_users
|
|
create_user("alice", "Password123!", role="admin", vaults=["*"])
|
|
assert has_users() is True
|
|
user = get_user("alice")
|
|
assert user is not None
|
|
assert user["username"] == "alice"
|
|
assert user["role"] == "admin"
|
|
assert user["vaults"] == ["*"]
|
|
|
|
def test_create_duplicate_raises(self):
|
|
from backend.auth.user_store import create_user
|
|
create_user("bob", "Password123!")
|
|
with pytest.raises(ValueError, match="already exists"):
|
|
create_user("bob", "Different456!")
|
|
|
|
def test_get_all_users(self):
|
|
from backend.auth.user_store import create_user, get_all_users
|
|
create_user("alice", "Pass123!!!", role="admin", vaults=["*"])
|
|
create_user("bob", "Pass456!!!", role="user", vaults=["Vault1"])
|
|
users = get_all_users()
|
|
assert len(users) == 2
|
|
|
|
def test_update_user(self):
|
|
from backend.auth.user_store import create_user, update_user, get_user
|
|
create_user("alice", "Pass123!!!")
|
|
update_user("alice", {"display_name": "Alice Updated", "vaults": ["V1", "V2"]})
|
|
user = get_user("alice")
|
|
assert user["display_name"] == "Alice Updated"
|
|
assert user["vaults"] == ["V1", "V2"]
|
|
|
|
def test_delete_user(self):
|
|
from backend.auth.user_store import create_user, delete_user, get_user
|
|
create_user("charlie", "Pass123!!!")
|
|
# delete_user returns None on success, raises ValueError if not found
|
|
assert delete_user("charlie") is None
|
|
assert get_user("charlie") is None
|
|
|
|
def test_delete_nonexistent(self):
|
|
from backend.auth.user_store import delete_user
|
|
with pytest.raises(ValueError, match="not found"):
|
|
delete_user("ghost")
|
|
|
|
def test_toggle_active(self):
|
|
from backend.auth.user_store import create_user, update_user, get_user
|
|
create_user("dave", "Pass123!!!")
|
|
update_user("dave", {"active": False})
|
|
user = get_user("dave")
|
|
assert user["active"] is False
|
|
|
|
def test_login_failure_and_lockout(self):
|
|
from backend.auth.user_store import (
|
|
create_user,
|
|
record_login_failure,
|
|
is_locked,
|
|
record_login_success,
|
|
)
|
|
create_user("eve", "Pass123!!!")
|
|
# Fail 5 times
|
|
for _ in range(5):
|
|
record_login_failure("eve")
|
|
assert is_locked("eve") is True
|
|
# Unlock after success
|
|
record_login_success("eve")
|
|
assert is_locked("eve") is False
|
|
|
|
def test_change_password(self):
|
|
from backend.auth.user_store import create_user, get_user
|
|
from backend.auth.password import verify_password
|
|
create_user("frank", "OldPass123!")
|
|
# Update password
|
|
from backend.auth.user_store import update_user
|
|
from backend.auth.password import hash_password
|
|
update_user("frank", {"password_hash": hash_password("NewPass456!")})
|
|
user = get_user("frank")
|
|
assert verify_password("NewPass456!", user["password_hash"])
|
|
assert not verify_password("OldPass123!", user["password_hash"])
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Rate Limiting
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
class TestRateLimiter:
|
|
def test_initial_state(self):
|
|
from backend.ratelimit import get_status
|
|
status = get_status("192.168.1.1")
|
|
assert status["limited"] is False
|
|
assert status["failures"] == 0
|
|
|
|
def test_failures_increment(self):
|
|
from backend.ratelimit import record_failure, is_rate_limited, record_success
|
|
ip = "10.0.0.1"
|
|
for _ in range(10):
|
|
record_failure(ip)
|
|
assert is_rate_limited(ip) is True
|
|
|
|
# Clear
|
|
record_success(ip)
|
|
assert is_rate_limited(ip) is False
|
|
|
|
def test_record_success_clears(self):
|
|
from backend.ratelimit import record_failure, record_success, is_rate_limited
|
|
ip = "172.16.0.1"
|
|
record_failure(ip)
|
|
record_failure(ip)
|
|
record_success(ip)
|
|
assert is_rate_limited(ip) is False
|
|
|
|
def test_global_status(self):
|
|
from backend.ratelimit import get_status
|
|
status = get_status()
|
|
assert "tracked_ips" in status
|
|
assert "max_attempts" in status
|
|
assert "limited_ips" in status
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Middleware (dependency-based)
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
class TestMiddleware:
|
|
def test_is_auth_enabled_logic(self):
|
|
"""is_auth_enabled reads OBSIGATE_AUTH_ENABLED env var."""
|
|
from backend.auth.middleware import is_auth_enabled
|
|
# The function checks if env var != 'false'
|
|
# In our test env it may be set by fixture — test the logic
|
|
result = is_auth_enabled()
|
|
assert isinstance(result, bool)
|
|
|
|
def test_check_vault_access(self):
|
|
from backend.auth.middleware import check_vault_access
|
|
admin = {"vaults": ["*"]}
|
|
user = {"vaults": ["Vault1", "Vault2"]}
|
|
nobody = {"vaults": []}
|
|
|
|
assert check_vault_access("Vault1", admin) is True
|
|
assert check_vault_access("AnyVault", admin) is True
|
|
assert check_vault_access("Vault1", user) is True
|
|
assert check_vault_access("Vault3", user) is False
|
|
assert check_vault_access("Vault1", nobody) is False
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Insecure (auth-disabled) deployment guard — BUG-037
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
class TestInsecureAuthGuard:
|
|
def test_is_loopback_host(self):
|
|
from backend.auth.middleware import is_loopback_host
|
|
|
|
assert is_loopback_host(None) is True
|
|
assert is_loopback_host("127.0.0.1") is True
|
|
assert is_loopback_host("::1") is True
|
|
assert is_loopback_host("[::1]") is True
|
|
assert is_loopback_host("localhost") is True
|
|
assert is_loopback_host("0.0.0.0") is False
|
|
assert is_loopback_host("192.168.1.10") is False
|
|
|
|
def test_bind_host_from_argv(self):
|
|
from backend.auth.middleware import bind_host_from_argv
|
|
|
|
assert bind_host_from_argv(
|
|
["uvicorn", "backend.main:app", "--host", "0.0.0.0", "--port", "8080"]
|
|
) == "0.0.0.0"
|
|
assert bind_host_from_argv(["uvicorn", "app", "--host=127.0.0.1"]) == "127.0.0.1"
|
|
assert bind_host_from_argv(["uvicorn", "app"]) is None
|
|
|
|
def test_guard_refuses_public_bind_without_optin(self, monkeypatch):
|
|
from backend import main
|
|
|
|
monkeypatch.setenv("OBSIGATE_AUTH_ENABLED", "false")
|
|
monkeypatch.delenv("OBSIGATE_ALLOW_INSECURE", raising=False)
|
|
monkeypatch.setattr("sys.argv", ["uvicorn", "backend.main:app", "--host", "0.0.0.0"])
|
|
with pytest.raises(RuntimeError):
|
|
main._guard_insecure_auth()
|
|
|
|
def test_guard_allows_loopback(self, monkeypatch):
|
|
from backend import main
|
|
|
|
monkeypatch.setenv("OBSIGATE_AUTH_ENABLED", "false")
|
|
monkeypatch.delenv("OBSIGATE_ALLOW_INSECURE", raising=False)
|
|
monkeypatch.setattr("sys.argv", ["uvicorn", "backend.main:app", "--host", "127.0.0.1"])
|
|
main._guard_insecure_auth() # must not raise
|
|
|
|
def test_guard_allows_explicit_optin(self, monkeypatch):
|
|
from backend import main
|
|
|
|
monkeypatch.setenv("OBSIGATE_AUTH_ENABLED", "false")
|
|
monkeypatch.setenv("OBSIGATE_ALLOW_INSECURE", "true")
|
|
monkeypatch.setattr("sys.argv", ["uvicorn", "backend.main:app", "--host", "0.0.0.0"])
|
|
main._guard_insecure_auth() # must not raise
|
|
|
|
def test_guard_noop_when_auth_enabled(self, monkeypatch):
|
|
from backend import main
|
|
|
|
monkeypatch.setenv("OBSIGATE_AUTH_ENABLED", "true")
|
|
monkeypatch.setattr("sys.argv", ["uvicorn", "backend.main:app", "--host", "0.0.0.0"])
|
|
main._guard_insecure_auth() # must not raise
|