feat(auth): implement Keycloak authentication with RBAC and pagination
Some checks failed
ci/woodpecker/pr/lint Pipeline failed
ci/woodpecker/pr/test Pipeline was successful
ci/woodpecker/pr/type Pipeline was successful

Major changes:
- Add Keycloak integration via token introspection endpoint
- Implement RBAC system with roles: admin, user, guest
- Add role-based permissions for post operations
- Add pagination support (default limit: 10) to list endpoints
- Add published_only filter with admin-only override for unpublished posts

Security improvements:
- Remove hardcoded default secrets (SECRET_KEY, KEYCLOAK_CLIENT_SECRET)
- Update .env.example with proper security placeholders
- Add comprehensive RBAC unit tests

Infrastructure:
- Add httpx dependency for HTTP client
- Add KeycloakAuthClient with token caching (TTL: 60s)
- Add role-based dependencies (RequireAdmin, RequireUser, etc.)
- Update DI container with Keycloak provider

Endpoints updated:
- GET /posts: filter by published status (admin can see all)
- Add pagination params (limit, offset) to list endpoints
- Enforce RBAC on post operations

Tests:
- Add 16 auth infrastructure tests
- Add 13 RBAC role tests
- Update existing tests for new required settings

Breaking changes:
- SECRET_KEY and KEYCLOAK_CLIENT_SECRET now required (no defaults)
This commit is contained in:
2026-05-02 00:43:10 +03:00
parent ddab62a883
commit 184b95969c
20 changed files with 1461 additions and 99 deletions

View File

@@ -1,23 +1,57 @@
"""API test fixtures."""
from typing import AsyncGenerator
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from httpx import ASGITransport, AsyncClient
from app.infrastructure.auth.models import TokenInfo
from app.main import app_factory
@pytest.fixture
async def client() -> AsyncGenerator[AsyncClient, None]:
def mock_keycloak_client() -> MagicMock:
"""Create mock Keycloak client for testing."""
mock_client = AsyncMock()
mock_client.introspect_token.return_value = TokenInfo(
active=True,
user_id="test-user-id",
username="testuser",
email="test@example.com",
roles=["user"],
)
return mock_client
@pytest.fixture
async def client(mock_keycloak_client: MagicMock) -> AsyncGenerator[AsyncClient, None]:
"""Create async HTTP client for API testing."""
app = app_factory()
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
yield ac
with patch(
"app.presentation.api.deps.KeycloakAuthClient",
return_value=mock_keycloak_client,
):
app = app_factory()
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
yield ac
@pytest.fixture
def auth_headers() -> dict[str, str]:
"""Return mock authentication headers."""
return {"Authorization": "Bearer test_token", "X-User-Id": "user-123"}
return {"Authorization": "Bearer test_token"}
@pytest.fixture
def unauthorized_keycloak_client() -> MagicMock:
"""Create mock Keycloak client that returns invalid token."""
mock_client = AsyncMock()
mock_client.introspect_token.return_value = TokenInfo(
active=False,
user_id="",
username="",
email="",
roles=[],
)
return mock_client

View File

@@ -0,0 +1,123 @@
"""Tests for role-based access control."""
from app.domain.roles import (
ROLE_PERMISSIONS,
Permission,
Role,
get_effective_role,
has_permission,
)
class TestRole:
"""Test Role enum."""
def test_role_values(self) -> None:
"""Test role enum values."""
assert Role.ADMIN.value == "admin"
assert Role.USER.value == "user"
assert Role.GUEST.value == "guest"
def test_role_comparison(self) -> None:
"""Test role comparison."""
assert Role.ADMIN == Role.ADMIN
# USER and ADMIN are different enum values with different string values
assert Role.USER.value != Role.ADMIN.value # type: ignore[comparison-overlap]
class TestPermissions:
"""Test permission definitions."""
def test_permission_values(self) -> None:
"""Test permission constants."""
assert Permission.POST_CREATE == "post:create"
assert Permission.POST_READ == "post:read"
assert Permission.POST_READ_UNPUBLISHED == "post:read_unpublished"
assert Permission.POST_UPDATE == "post:update"
assert Permission.POST_DELETE == "post:delete"
assert Permission.POST_PUBLISH == "post:publish"
class TestRolePermissions:
"""Test role-based permission mapping."""
def test_admin_has_all_permissions(self) -> None:
"""Test admin has all permissions."""
admin_perms = ROLE_PERMISSIONS[Role.ADMIN]
assert Permission.POST_CREATE in admin_perms
assert Permission.POST_READ in admin_perms
assert Permission.POST_READ_UNPUBLISHED in admin_perms
assert Permission.POST_UPDATE in admin_perms
assert Permission.POST_DELETE in admin_perms
assert Permission.POST_PUBLISH in admin_perms
def test_user_permissions(self) -> None:
"""Test user permissions."""
user_perms = ROLE_PERMISSIONS[Role.USER]
assert Permission.POST_CREATE in user_perms
assert Permission.POST_READ in user_perms
assert Permission.POST_UPDATE in user_perms
assert Permission.POST_DELETE in user_perms
assert Permission.POST_PUBLISH in user_perms
# User cannot read unpublished
assert Permission.POST_READ_UNPUBLISHED not in user_perms
def test_guest_permissions(self) -> None:
"""Test guest permissions."""
guest_perms = ROLE_PERMISSIONS[Role.GUEST]
assert Permission.POST_READ in guest_perms
# Guest has very limited permissions
assert Permission.POST_CREATE not in guest_perms
assert Permission.POST_UPDATE not in guest_perms
assert Permission.POST_DELETE not in guest_perms
assert Permission.POST_READ_UNPUBLISHED not in guest_perms
class TestHasPermission:
"""Test has_permission function."""
def test_admin_has_all_permissions_check(self) -> None:
"""Test admin permission checks."""
assert has_permission(Role.ADMIN, Permission.POST_CREATE) is True
assert has_permission(Role.ADMIN, Permission.POST_READ_UNPUBLISHED) is True
assert has_permission(Role.ADMIN, "unknown:permission") is False
def test_user_limited_permissions(self) -> None:
"""Test user limited permissions."""
assert has_permission(Role.USER, Permission.POST_CREATE) is True
assert has_permission(Role.USER, Permission.POST_READ_UNPUBLISHED) is False
assert has_permission(Role.USER, Permission.POST_READ) is True
def test_guest_read_only(self) -> None:
"""Test guest read-only access."""
assert has_permission(Role.GUEST, Permission.POST_READ) is True
assert has_permission(Role.GUEST, Permission.POST_CREATE) is False
assert has_permission(Role.GUEST, Permission.POST_UPDATE) is False
class TestGetEffectiveRole:
"""Test get_effective_role function."""
def test_admin_from_roles_list(self) -> None:
"""Test admin role detection."""
assert get_effective_role(["admin"]) == Role.ADMIN
assert get_effective_role(["user", "admin"]) == Role.ADMIN
assert get_effective_role(["admin", "user"]) == Role.ADMIN
def test_user_from_roles_list(self) -> None:
"""Test user role detection."""
assert get_effective_role(["user"]) == Role.USER
assert get_effective_role(["user", "moderator"]) == Role.USER
def test_guest_from_roles_list(self) -> None:
"""Test guest role detection."""
assert get_effective_role([]) == Role.GUEST
assert get_effective_role(["unknown"]) == Role.GUEST
assert get_effective_role(["guest"]) == Role.GUEST
def test_role_priority(self) -> None:
"""Test that admin > user > guest."""
# Admin takes precedence
assert get_effective_role(["user", "admin", "guest"]) == Role.ADMIN
# User takes precedence over guest
assert get_effective_role(["guest", "user"]) == Role.USER

View File

@@ -0,0 +1,318 @@
"""Tests for Keycloak authentication client."""
from unittest.mock import AsyncMock, Mock, patch
import pytest
from app.infrastructure.auth import KeycloakAuthClient, KeycloakUser, TokenInfo
from app.infrastructure.config.settings import Settings
class TestTokenInfo:
"""Test TokenInfo dataclass."""
def test_token_info_valid(self) -> None:
"""Test valid token info."""
token_info = TokenInfo(
active=True,
user_id="user-123",
username="testuser",
email="test@example.com",
roles=["user"],
)
assert token_info.is_valid is True
assert token_info.user_id == "user-123"
assert token_info.username == "testuser"
assert token_info.email == "test@example.com"
assert token_info.roles == ["user"]
def test_token_info_invalid_not_active(self) -> None:
"""Test invalid token when not active."""
token_info = TokenInfo(
active=False,
user_id="user-123",
username="testuser",
email="test@example.com",
roles=["user"],
)
assert token_info.is_valid is False
def test_token_info_invalid_no_user_id(self) -> None:
"""Test invalid token when no user_id."""
token_info = TokenInfo(
active=True,
user_id="",
username="testuser",
email="test@example.com",
roles=["user"],
)
assert token_info.is_valid is False
def test_token_info_empty_roles(self) -> None:
"""Test token info with empty roles."""
token_info = TokenInfo(
active=True,
user_id="user-123",
username="testuser",
email="test@example.com",
roles=[],
)
assert token_info.is_valid is True
assert token_info.roles == []
class TestKeycloakUser:
"""Test KeycloakUser dataclass."""
def test_keycloak_user_creation(self) -> None:
"""Test KeycloakUser creation."""
user = KeycloakUser(
id="user-123",
username="testuser",
email="test@example.com",
first_name="Test",
last_name="User",
roles=["user", "admin"],
is_active=True,
)
assert user.id == "user-123"
assert user.username == "testuser"
assert user.email == "test@example.com"
assert user.first_name == "Test"
assert user.last_name == "User"
assert user.roles == ["user", "admin"]
assert user.is_active is True
def test_keycloak_user_defaults(self) -> None:
"""Test KeycloakUser with default values."""
user = KeycloakUser(
id="user-123",
username="testuser",
email="test@example.com",
)
assert user.first_name == ""
assert user.last_name == ""
assert user.roles == []
assert user.is_active is True
class TestKeycloakAuthClient:
"""Test KeycloakAuthClient."""
@pytest.fixture
def settings(self) -> Settings:
"""Create test settings."""
from app.infrastructure.config import KCConfig, SecurityConfig
return Settings(
environment="dev",
kc=KCConfig(
server_url="http://localhost:8080",
realm="test-realm",
client_id="test-client",
client_secret="test-secret",
token_cache_ttl=60,
),
security=SecurityConfig(
secret_key="test-secret-key-for-jwt-tokens",
),
)
@pytest.fixture
def client(self, settings: Settings) -> KeycloakAuthClient:
"""Create Keycloak client."""
return KeycloakAuthClient(settings)
def test_client_initialization(
self, client: KeycloakAuthClient, settings: Settings
) -> None:
"""Test client initialization."""
assert client._settings == settings
assert client._base_url == "http://localhost:8080/realms/test-realm"
assert client._client_id == "test-client"
assert client._client_secret == "test-secret"
assert client._cache_ttl == 60
def test_get_introspection_url(self, client: KeycloakAuthClient) -> None:
"""Test introspection URL generation."""
url = client._get_introspection_url()
assert (
url
== "http://localhost:8080/realms/test-realm/protocol/openid-connect/token/introspection"
)
def test_get_userinfo_url(self, client: KeycloakAuthClient) -> None:
"""Test userinfo URL generation."""
url = client._get_userinfo_url()
assert (
url
== "http://localhost:8080/realms/test-realm/protocol/openid-connect/userinfo"
)
@pytest.mark.asyncio
async def test_introspect_token_success(self, client: KeycloakAuthClient) -> None:
"""Test successful token introspection."""
mock_response = Mock()
mock_response.json.return_value = {
"active": True,
"sub": "user-123",
"preferred_username": "testuser",
"email": "test@example.com",
"realm_access": {"roles": ["user", "admin"]},
}
mock_response.raise_for_status = Mock()
mock_async_client = AsyncMock()
mock_async_client.__aenter__ = AsyncMock(return_value=mock_async_client)
mock_async_client.__aexit__ = AsyncMock(return_value=None)
mock_async_client.post = AsyncMock(return_value=mock_response)
with patch("httpx.AsyncClient", return_value=mock_async_client):
result = await client.introspect_token("test-token")
assert result.active is True
assert result.user_id == "user-123"
assert result.username == "testuser"
assert result.email == "test@example.com"
assert result.roles == ["user", "admin"]
assert result.is_valid is True
@pytest.mark.asyncio
async def test_introspect_token_inactive(self, client: KeycloakAuthClient) -> None:
"""Test introspection with inactive token."""
mock_response = Mock()
mock_response.json.return_value = {"active": False}
mock_response.raise_for_status = Mock()
mock_async_client = AsyncMock()
mock_async_client.__aenter__ = AsyncMock(return_value=mock_async_client)
mock_async_client.__aexit__ = AsyncMock(return_value=None)
mock_async_client.post = AsyncMock(return_value=mock_response)
with patch("httpx.AsyncClient", return_value=mock_async_client):
result = await client.introspect_token("test-token")
assert result.active is False
assert result.is_valid is False
@pytest.mark.asyncio
async def test_introspect_token_http_error(
self, client: KeycloakAuthClient
) -> None:
"""Test introspection with HTTP error."""
import httpx
mock_async_client = AsyncMock()
mock_async_client.__aenter__ = AsyncMock(return_value=mock_async_client)
mock_async_client.__aexit__ = AsyncMock(return_value=None)
mock_async_client.post = AsyncMock(
side_effect=httpx.HTTPError("Connection error")
)
with patch("httpx.AsyncClient", return_value=mock_async_client):
result = await client.introspect_token("test-token")
assert result.active is False
assert result.is_valid is False
@pytest.mark.asyncio
async def test_introspect_token_uses_cache(
self, client: KeycloakAuthClient
) -> None:
"""Test that token introspection uses cache."""
mock_response = Mock()
mock_response.json.return_value = {
"active": True,
"sub": "user-123",
"preferred_username": "testuser",
"email": "test@example.com",
"realm_access": {"roles": ["user"]},
}
mock_response.raise_for_status = Mock()
mock_async_client = AsyncMock()
mock_async_client.__aenter__ = AsyncMock(return_value=mock_async_client)
mock_async_client.__aexit__ = AsyncMock(return_value=None)
mock_async_client.post = AsyncMock(return_value=mock_response)
with patch("httpx.AsyncClient", return_value=mock_async_client):
# First call
result1 = await client.introspect_token("test-token")
# Second call should use cache
result2 = await client.introspect_token("test-token")
# HTTP client should only be called once
assert mock_async_client.post.call_count == 1
assert result1.user_id == result2.user_id
@pytest.mark.asyncio
async def test_get_userinfo_success(self, client: KeycloakAuthClient) -> None:
"""Test successful userinfo retrieval."""
mock_response = Mock()
mock_response.json.return_value = {
"sub": "user-123",
"preferred_username": "testuser",
"email": "test@example.com",
"given_name": "Test",
"family_name": "User",
"realm_access": {"roles": ["user"]},
}
mock_response.raise_for_status = Mock()
mock_async_client = AsyncMock()
mock_async_client.__aenter__ = AsyncMock(return_value=mock_async_client)
mock_async_client.__aexit__ = AsyncMock(return_value=None)
mock_async_client.get = AsyncMock(return_value=mock_response)
with patch("httpx.AsyncClient", return_value=mock_async_client):
result = await client.get_userinfo("test-token")
assert result is not None
assert result.id == "user-123"
assert result.username == "testuser"
assert result.email == "test@example.com"
assert result.first_name == "Test"
assert result.last_name == "User"
assert result.roles == ["user"]
@pytest.mark.asyncio
async def test_get_userinfo_error(self, client: KeycloakAuthClient) -> None:
"""Test userinfo retrieval with error."""
import httpx
mock_async_client = AsyncMock()
mock_async_client.__aenter__ = AsyncMock(return_value=mock_async_client)
mock_async_client.__aexit__ = AsyncMock(return_value=None)
mock_async_client.get = AsyncMock(
side_effect=httpx.HTTPError("Connection error")
)
with patch("httpx.AsyncClient", return_value=mock_async_client):
result = await client.get_userinfo("test-token")
assert result is None
@pytest.mark.asyncio
async def test_introspect_token_no_realm_roles(
self, client: KeycloakAuthClient
) -> None:
"""Test introspection without realm_access roles."""
mock_response = Mock()
mock_response.json.return_value = {
"active": True,
"sub": "user-123",
"preferred_username": "testuser",
"email": "test@example.com",
}
mock_response.raise_for_status = Mock()
mock_async_client = AsyncMock()
mock_async_client.__aenter__ = AsyncMock(return_value=mock_async_client)
mock_async_client.__aexit__ = AsyncMock(return_value=None)
mock_async_client.post = AsyncMock(return_value=mock_response)
with patch("httpx.AsyncClient", return_value=mock_async_client):
result = await client.introspect_token("test-token")
assert result.active is True
assert result.roles == []

View File

@@ -1,37 +1,247 @@
"""Tests for infrastructure config."""
from app.infrastructure.config import Settings
import pytest
from app.infrastructure.config import (
AppConfig,
DBConfig,
Environment,
KCConfig,
SecurityConfig,
Settings,
)
class TestSettings:
"""Test Settings with composition pattern."""
def test_default_values(self) -> None:
"""Test default settings values by creating settings without env file."""
# Create settings with no env file to test defaults
s = Settings(_env_file=None)
assert s.app_name == "Blog API"
assert s.debug is False
assert s.host == "0.0.0.0"
assert s.port == 8000
assert s.database_url == "sqlite:///./blog.db"
assert s.database_echo is False
# Create settings with required secrets and no env file
s = Settings(
_env_file=None,
security=SecurityConfig(secret_key="test-secret-key"),
kc=KCConfig(client_secret="test-client-secret"),
)
assert s.app.name == "Blog API"
assert s.app.debug is False
assert s.app.host == "0.0.0.0"
assert s.app.port == 8000
assert s.database_url == "sqlite+aiosqlite:///./blog.db"
assert s.db.echo is False
assert s.security.secret_key == "test-secret-key"
assert s.kc.client_secret == "test-client-secret"
assert s.environment == Environment.DEV
def test_custom_values(self) -> None:
"""Test custom settings values."""
s = Settings(
app_name="Test API",
debug=True,
host="localhost",
port=9000,
database_url="postgresql://test",
secret_key="test-secret",
_env_file=None,
environment=Environment.PROD,
app=AppConfig(
name="Test API",
debug=True,
host="localhost",
port=9000,
),
db=DBConfig(url="postgresql+asyncpg://user:pass@host/db"),
security=SecurityConfig(secret_key="test-secret"),
kc=KCConfig(client_secret="test-client-secret"),
)
assert s.app_name == "Test API"
assert s.debug is True
assert s.host == "localhost"
assert s.port == 9000
assert s.database_url == "postgresql://test"
assert s.secret_key == "test-secret"
assert s.app.name == "Test API"
assert s.app.debug is True
assert s.app.host == "localhost"
assert s.app.port == 9000
assert s.database_url == "postgresql+asyncpg://user:pass@host/db"
assert s.security.secret_key == "test-secret"
assert s.kc.client_secret == "test-client-secret"
assert s.environment == Environment.PROD
def test_model_config(self) -> None:
"""Test settings model config."""
assert "env_file" in Settings.model_config
def test_is_dev_property(self) -> None:
"""Test is_dev property."""
s = Settings(
_env_file=None,
environment=Environment.DEV,
security=SecurityConfig(secret_key="test"),
kc=KCConfig(client_secret="test"),
)
assert s.is_dev is True
assert s.is_prod is False
def test_is_prod_property(self) -> None:
"""Test is_prod property."""
s = Settings(
_env_file=None,
environment=Environment.PROD,
security=SecurityConfig(secret_key="test"),
kc=KCConfig(client_secret="test"),
)
assert s.is_prod is True
assert s.is_dev is False
def test_prod_requires_security_secret(self) -> None:
"""Test that prod mode requires security secret_key."""
with pytest.raises(ValueError, match="SECURITY_SECRET_KEY"):
Settings(
_env_file=None,
environment=Environment.PROD,
security=SecurityConfig(secret_key=""),
kc=KCConfig(client_secret="test"),
)
def test_prod_requires_kc_secret(self) -> None:
"""Test that prod mode requires KC client_secret."""
with pytest.raises(ValueError, match="KC_CLIENT_SECRET"):
Settings(
_env_file=None,
environment=Environment.PROD,
security=SecurityConfig(secret_key="test"),
kc=KCConfig(client_secret=""),
)
def test_database_url_dev_default(self) -> None:
"""Test default database URL in dev mode."""
s = Settings(
_env_file=None,
environment=Environment.DEV,
security=SecurityConfig(secret_key="test"),
kc=KCConfig(client_secret="test"),
)
assert s.database_url == "sqlite+aiosqlite:///./blog.db"
def test_database_url_prod_builds_postgres(self) -> None:
"""Test that database URL builds from components in prod."""
s = Settings(
_env_file=None,
environment=Environment.PROD,
db=DBConfig(
url=None, # Force building from components
host="db.example.com",
port=5433,
user="admin",
password="secret",
name="mydb",
),
security=SecurityConfig(secret_key="test"),
kc=KCConfig(client_secret="test"),
)
assert (
s.database_url
== "postgresql+asyncpg://admin:secret@db.example.com:5433/mydb"
)
def test_database_url_override(self) -> None:
"""Test that explicit database URL overrides auto-building."""
s = Settings(
_env_file=None,
environment=Environment.PROD,
db=DBConfig(
url="postgresql+asyncpg://custom/url",
host="ignored",
user="ignored",
),
security=SecurityConfig(secret_key="test"),
kc=KCConfig(client_secret="test"),
)
assert s.database_url == "postgresql+asyncpg://custom/url"
class TestAppConfig:
"""Test AppConfig."""
def test_default_values(self) -> None:
"""Test AppConfig default values."""
cfg = AppConfig()
assert cfg.name == "Blog API"
assert cfg.debug is False
assert cfg.host == "0.0.0.0"
assert cfg.port == 8000
class TestDBConfig:
"""Test DBConfig."""
def test_default_values(self) -> None:
"""Test DBConfig default values."""
cfg = DBConfig()
assert cfg.url is None
assert cfg.echo is False
assert cfg.host == "localhost"
assert cfg.port == 5432
assert cfg.user == "postgres"
assert cfg.password == "postgres"
assert cfg.name == "blog"
def test_postgres_url_validation(self) -> None:
"""Test URL validation for postgres."""
cfg = DBConfig(url="postgresql+asyncpg://user:pass@host/db")
assert cfg.url == "postgresql+asyncpg://user:pass@host/db"
def test_sqlite_url_validation(self) -> None:
"""Test URL validation for sqlite."""
cfg = DBConfig(url="sqlite+aiosqlite:///./test.db")
assert cfg.url == "sqlite+aiosqlite:///./test.db"
def test_invalid_url_validation(self) -> None:
"""Test URL validation rejects invalid URLs."""
with pytest.raises(ValueError, match="sqlite+.*postgresql+"):
DBConfig(url="mysql://invalid")
class TestKCConfig:
"""Test KCConfig."""
def test_default_values(self) -> None:
"""Test KCConfig default values."""
cfg = KCConfig(client_secret="test-secret")
assert cfg.server_url == "http://localhost:8080"
assert cfg.realm == "blog"
assert cfg.client_id == "blog-api"
assert cfg.client_secret == "test-secret"
assert cfg.token_cache_ttl == 60
def test_is_configured_with_secret(self) -> None:
"""Test is_configured returns True when secret is set."""
cfg = KCConfig(client_secret="test-secret")
assert cfg.is_configured is True
def test_is_configured_without_secret(self) -> None:
"""Test is_configured returns False when secret is empty."""
cfg = KCConfig(client_secret="")
assert cfg.is_configured is False
class TestSecurityConfig:
"""Test SecurityConfig."""
def test_default_values(self) -> None:
"""Test SecurityConfig default values."""
cfg = SecurityConfig(secret_key="test-key")
assert cfg.secret_key == "test-key"
assert cfg.access_token_expire_minutes == 30
def test_is_configured_with_secret(self) -> None:
"""Test is_configured returns True when secret is set."""
cfg = SecurityConfig(secret_key="test-secret")
assert cfg.is_configured is True
def test_is_configured_without_secret(self) -> None:
"""Test is_configured returns False when secret is empty."""
cfg = SecurityConfig(secret_key="")
assert cfg.is_configured is False
class TestEnvironment:
"""Test Environment enum."""
def test_dev_value(self) -> None:
"""Test DEV environment value."""
assert Environment.DEV.value == "dev"
def test_prod_value(self) -> None:
"""Test PROD environment value."""
assert Environment.PROD.value == "prod"