infrasynth-backend-kit/tests/test_security/test_auth.py

from datetime import timedelta

import pytest
from django.contrib.auth import get_user_model
from django.contrib.auth.hashers import make_password
from django.utils import timezone
from rest_framework import status
from rest_framework.test import APIClient, APIRequestFactory

from infrasynth.security.auth.api_keys import APIKeyAuthentication, SystemUser
from infrasynth.security.auth.backends import EmailOrUsernameBackend
from infrasynth.security.auth.cookies import CookieJWTAuthentication
from infrasynth.security.models import APIKey
from infrasynth.shared.crypto import encrypt

UserModel = get_user_model()


class TestEmailOrUsernameBackend:
    def test_authenticates_by_username(self, user):
        backend = EmailOrUsernameBackend()
        result = backend.authenticate(None, username="testuser", password="testpass123")
        assert result == user

    def test_authenticates_by_email(self, user):
        backend = EmailOrUsernameBackend()
        result = backend.authenticate(None, username="test@example.com", password="testpass123")
        assert result == user

    def test_wrong_password_returns_none(self, user):
        backend = EmailOrUsernameBackend()
        assert backend.authenticate(None, username="testuser", password="wrong") is None

    def test_unknown_user_returns_none(self, user):
        backend = EmailOrUsernameBackend()
        assert backend.authenticate(None, username="ghost", password="testpass123") is None

    def test_inactive_user_returns_none(self, db):
        UserModel.objects.create_user(
            username="inactive", email="inactive@example.com", password="testpass123", is_active=False
        )
        backend = EmailOrUsernameBackend()
        assert backend.authenticate(None, username="inactive", password="testpass123") is None

    def test_missing_credentials_returns_none(self, user):
        backend = EmailOrUsernameBackend()
        assert backend.authenticate(None, username=None, password=None) is None


class TestCookieJWTAuthentication:
    def test_authenticates_with_valid_cookie(self, user):
        from rest_framework_simplejwt.tokens import RefreshToken

        token = str(RefreshToken.for_user(user).access_token)
        encrypted = encrypt(token)
        request = APIRequestFactory().get("/")
        request.COOKIES["access_token"] = encrypted

        auth = CookieJWTAuthentication()
        authenticated_user, validated = auth.authenticate(request)
        assert authenticated_user == user
        assert validated

    def test_no_cookie_returns_none(self, user):
        request = APIRequestFactory().get("/")
        auth = CookieJWTAuthentication()
        assert auth.authenticate(request) is None

    def test_invalid_cookie_raises(self, user):
        from rest_framework.exceptions import AuthenticationFailed

        request = APIRequestFactory().get("/")
        request.COOKIES["access_token"] = "not-a-valid-token"
        auth = CookieJWTAuthentication()
        with pytest.raises(AuthenticationFailed):
            auth.authenticate(request)


class TestAPIKeyAuthentication:
    @pytest.fixture
    def api_key(self, db, user):
        secret = "s3cr3t-api-secret"
        prefix = "testkey1"
        return APIKey.objects.create(
            name="Test Key",
            prefix=prefix,
            key_hash=make_password(secret),
            scopes=["read:users", "write:billing"],
            created_by=user,
        )

    def test_authenticates_with_valid_key(self, api_key):
        request = APIRequestFactory().get("/", HTTP_X_API_KEY="testkey1.s3cr3t-api-secret")
        auth = APIKeyAuthentication()
        auth_user, key = auth.authenticate(request)
        assert isinstance(auth_user, SystemUser)
        assert auth_user.scopes == ["read:users", "write:billing"]
        assert key == api_key

    def test_no_header_returns_none(self, api_key):
        request = APIRequestFactory().get("/")
        auth = APIKeyAuthentication()
        assert auth.authenticate(request) is None

    def test_invalid_format_raises(self, api_key):
        from rest_framework.exceptions import AuthenticationFailed

        request = APIRequestFactory().get("/", HTTP_X_API_KEY="no-dot-here")
        auth = APIKeyAuthentication()
        with pytest.raises(AuthenticationFailed):
            auth.authenticate(request)

    def test_unknown_prefix_raises(self, api_key):
        from rest_framework.exceptions import AuthenticationFailed

        request = APIRequestFactory().get("/", HTTP_X_API_KEY="unknownkey.secret")
        auth = APIKeyAuthentication()
        with pytest.raises(AuthenticationFailed):
            auth.authenticate(request)

    def test_wrong_secret_raises(self, api_key):
        from rest_framework.exceptions import AuthenticationFailed

        request = APIRequestFactory().get("/", HTTP_X_API_KEY="testkey1.wrongsecret")
        auth = APIKeyAuthentication()
        with pytest.raises(AuthenticationFailed):
            auth.authenticate(request)

    def test_inactive_key_raises(self, api_key):
        from rest_framework.exceptions import AuthenticationFailed

        api_key.is_active = False
        api_key.save()
        request = APIRequestFactory().get("/", HTTP_X_API_KEY="testkey1.s3cr3t-api-secret")
        auth = APIKeyAuthentication()
        with pytest.raises(AuthenticationFailed):
            auth.authenticate(request)

    def test_expired_key_raises(self, api_key):
        from rest_framework.exceptions import AuthenticationFailed

        api_key.expires_at = timezone.now() - timedelta(days=1)
        api_key.save()
        request = APIRequestFactory().get("/", HTTP_X_API_KEY="testkey1.s3cr3t-api-secret")
        auth = APIKeyAuthentication()
        with pytest.raises(AuthenticationFailed):
            auth.authenticate(request)

    def test_updates_last_used_at(self, api_key):
        request = APIRequestFactory().get("/", HTTP_X_API_KEY="testkey1.s3cr3t-api-secret")
        auth = APIKeyAuthentication()
        auth.authenticate(request)
        api_key.refresh_from_db()
        assert api_key.last_used_at is not None


class TestSystemUser:
    def test_scopes_and_auth_flags(self):
        sys_user = SystemUser(scopes=["read:x"])
        assert sys_user.is_authenticated is True
        assert sys_user.is_anonymous is False
        assert sys_user.is_superuser is False
        assert sys_user.scopes == ["read:x"]

    def test_default_scopes(self):
        assert SystemUser().scopes == []


class TestAuthFlow:
    def test_login_sets_cookies(self, db, user):
        client = APIClient()
        resp = client.post(
            "/api/v1/auth/login/",
            {"username": "testuser", "password": "testpass123"},
            format="json",
        )
        assert resp.status_code == status.HTTP_200_OK
        assert "access_token" in resp.cookies
        assert "refresh_token" in resp.cookies

    def test_login_by_email(self, db, user):
        client = APIClient()
        resp = client.post(
            "/api/v1/auth/login/",
            {"username": "test@example.com", "password": "testpass123"},
            format="json",
        )
        assert resp.status_code == status.HTTP_200_OK

    def test_login_invalid_credentials(self, db, user):
        client = APIClient()
        resp = client.post(
            "/api/v1/auth/login/",
            {"username": "testuser", "password": "wrongpass"},
            format="json",
        )
        assert resp.status_code == status.HTTP_401_UNAUTHORIZED

    def test_login_inactive_user(self, db):
        UserModel.objects.create_user(
            username="banned", email="banned@example.com", password="testpass123", is_active=False
        )
        client = APIClient()
        resp = client.post("/api/v1/auth/login/", {"username": "banned", "password": "testpass123"}, format="json")
        assert resp.status_code == status.HTTP_401_UNAUTHORIZED

    def test_check_authenticated(self, authenticated_client, user):
        resp = authenticated_client.get("/api/v1/auth/check/")
        assert resp.status_code == status.HTTP_200_OK
        data = resp.json()
        assert data["id"] == user.pk
        assert data["email"] == user.email
        assert "effective_permissions" in data

    def test_check_superuser_has_wildcard(self, admin_client, admin_user):
        resp = admin_client.get("/api/v1/auth/check/")
        data = resp.json()
        assert data["effective_permissions"] == ["*"]

    def test_check_unauthenticated(self, api_client):
        resp = api_client.get("/api/v1/auth/check/")
        assert resp.status_code == status.HTTP_401_UNAUTHORIZED
        assert resp.json() == {"authenticated": False}

    def test_refresh_with_valid_cookie(self, db, user):
        from rest_framework_simplejwt.tokens import RefreshToken

        refresh = str(RefreshToken.for_user(user))
        client = APIClient()
        client.cookies["refresh_token"] = encrypt(refresh)
        resp = client.post("/api/v1/auth/refresh/")
        assert resp.status_code == status.HTTP_200_OK
        assert "access_token" in resp.cookies

    def test_refresh_without_cookie(self, api_client):
        resp = api_client.post("/api/v1/auth/refresh/")
        assert resp.status_code == status.HTTP_401_UNAUTHORIZED

    def test_refresh_invalid_cookie(self, api_client):
        api_client.cookies["refresh_token"] = "garbage-token"
        resp = api_client.post("/api/v1/auth/refresh/")
        assert resp.status_code == status.HTTP_401_UNAUTHORIZED

    def test_logout_clears_cookies(self, authenticated_client):
        resp = authenticated_client.post("/api/v1/auth/logout/")
        assert resp.status_code == status.HTTP_200_OK
        assert resp.cookies["access_token"].value == ""
        assert resp.cookies["refresh_token"].value == ""

    @pytest.mark.parametrize("crypto_key", [None, ""])
    def test_login_falls_back_to_plaintext_tokens(self, db, user, settings, crypto_key):
        settings.INFRASYNTH_SECURITY = {
            **settings.INFRASYNTH_SECURITY,
            "CRYPTO_KEY": crypto_key,
        }
        client = APIClient()
        resp = client.post(
            "/api/v1/auth/login/",
            {"username": "testuser", "password": "testpass123"},
            format="json",
        )
        assert resp.status_code == status.HTTP_200_OK
        assert resp.cookies["access_token"].value

    def test_full_login_roundtrip(self, db, user):
        client = APIClient()
        resp = client.post(
            "/api/v1/auth/login/",
            {"username": "testuser", "password": "testpass123"},
            format="json",
        )
        assert resp.status_code == status.HTTP_200_OK
        check = client.get("/api/v1/auth/check/")
        assert check.status_code == status.HTTP_200_OK
        assert check.json()["id"] == user.pk