feat(auth): sync Keycloak users for authorization
This commit is contained in:
@@ -37,11 +37,17 @@ but still need a project membership for project business APIs.
|
||||
Every authenticated metadata-user resolution validates the Keycloak access token
|
||||
and reads `sub`, `preferred_username`, and `email` claims. The
|
||||
backend finds `users` by `keycloak_id = sub`, rejects inactive or missing users,
|
||||
then refreshes `username`, `email`, and `last_login_at`.
|
||||
then refreshes `username`, `email`, and `last_login_at` for active users.
|
||||
|
||||
This keeps local display data current without changing the identity binding.
|
||||
There is no Keycloak webhook requirement; second-level user or permission sync is
|
||||
out of scope unless explicitly requested later.
|
||||
System administrators explicitly synchronize the Keycloak user directory from
|
||||
the system-management page. New identities are stored with role `user` and
|
||||
`is_active = false`; existing identities keep their TJWater role, active state,
|
||||
superuser flag, and project memberships. This makes pending accounts visible
|
||||
without requiring the user to log in and without granting access automatically.
|
||||
The administrator's Keycloak account must have the `realm-management`
|
||||
`query-users` and `view-users` client roles. The backend forwards that access
|
||||
token only to the same token issuer's Keycloak Admin API; no management
|
||||
credential is exposed to the browser.
|
||||
|
||||
## Admin APIs
|
||||
|
||||
|
||||
@@ -12,11 +12,17 @@ from app.auth.metadata_dependencies import (
|
||||
get_current_metadata_admin,
|
||||
get_metadata_repository,
|
||||
)
|
||||
from app.auth.keycloak_dependencies import (
|
||||
get_current_keycloak_payload,
|
||||
oauth2_optional,
|
||||
)
|
||||
from app.core.audit import AuditAction, log_audit_event
|
||||
from app.domain.schemas.admin_metadata import (
|
||||
AdminProjectCreateRequest,
|
||||
AdminProjectResponse,
|
||||
AdminProjectUpdateRequest,
|
||||
KeycloakUserSyncIssue,
|
||||
KeycloakUserSyncResponse,
|
||||
MetadataUsersBatchSyncRequest,
|
||||
MetadataUserResponse,
|
||||
MetadataUserSyncRequest,
|
||||
@@ -36,6 +42,11 @@ from app.infra.db.metadb.repositories.metadata_repository import (
|
||||
MetadataRepository,
|
||||
ProjectDbRouting,
|
||||
)
|
||||
from app.services.keycloak_user_sync import (
|
||||
KeycloakUserSyncConfigurationError,
|
||||
KeycloakUserSyncRequestError,
|
||||
fetch_keycloak_users,
|
||||
)
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
@@ -229,6 +240,94 @@ async def sync_metadata_users_batch(
|
||||
return results
|
||||
|
||||
|
||||
@router.post(
|
||||
"/admin/keycloak-user-syncs",
|
||||
response_model=KeycloakUserSyncResponse,
|
||||
)
|
||||
async def sync_keycloak_user_directory(
|
||||
current_user=Depends(get_current_metadata_admin),
|
||||
keycloak_payload: dict = Depends(get_current_keycloak_payload),
|
||||
keycloak_access_token: str | None = Depends(oauth2_optional),
|
||||
metadata_repo: MetadataRepository = Depends(get_metadata_repository),
|
||||
) -> KeycloakUserSyncResponse:
|
||||
try:
|
||||
keycloak_users = await fetch_keycloak_users(
|
||||
issuer=str(keycloak_payload.get("iss") or ""),
|
||||
bearer_token=keycloak_access_token or "",
|
||||
)
|
||||
except KeycloakUserSyncConfigurationError as exc:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||
detail=str(exc),
|
||||
) from exc
|
||||
except KeycloakUserSyncRequestError as exc:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||||
detail=str(exc),
|
||||
) from exc
|
||||
|
||||
created = 0
|
||||
updated = 0
|
||||
issues: list[KeycloakUserSyncIssue] = []
|
||||
for keycloak_user in keycloak_users:
|
||||
if keycloak_user.service_account_client_id:
|
||||
issues.append(
|
||||
KeycloakUserSyncIssue(
|
||||
username=keycloak_user.username,
|
||||
reason="service_account",
|
||||
)
|
||||
)
|
||||
continue
|
||||
if not keycloak_user.email:
|
||||
issues.append(
|
||||
KeycloakUserSyncIssue(
|
||||
username=keycloak_user.username,
|
||||
reason="missing_email",
|
||||
)
|
||||
)
|
||||
continue
|
||||
try:
|
||||
_, was_created = await metadata_repo.sync_user_snapshot_from_keycloak(
|
||||
keycloak_id=keycloak_user.keycloak_id,
|
||||
username=keycloak_user.username,
|
||||
email=keycloak_user.email,
|
||||
)
|
||||
except IntegrityError:
|
||||
await metadata_repo.session.rollback()
|
||||
issues.append(
|
||||
KeycloakUserSyncIssue(
|
||||
username=keycloak_user.username,
|
||||
reason="identity_conflict",
|
||||
)
|
||||
)
|
||||
except SQLAlchemyError as exc:
|
||||
await metadata_repo.session.rollback()
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||
detail="Metadata database error during Keycloak user synchronization",
|
||||
) from exc
|
||||
else:
|
||||
created += int(was_created)
|
||||
updated += int(not was_created)
|
||||
|
||||
result = KeycloakUserSyncResponse(
|
||||
scanned=len(keycloak_users),
|
||||
created=created,
|
||||
updated=updated,
|
||||
skipped=len(issues),
|
||||
issues=issues,
|
||||
)
|
||||
await log_audit_event(
|
||||
action=AuditAction.UPDATE,
|
||||
user_id=current_user.id,
|
||||
resource_type="keycloak_user_directory",
|
||||
request_data=result.model_dump(mode="json", exclude={"issues"}),
|
||||
response_status=status.HTTP_200_OK,
|
||||
session=metadata_repo.session,
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
@router.get("/admin/users", response_model=List[MetadataUserResponse])
|
||||
async def list_metadata_users(
|
||||
skip: int = Query(0, ge=0),
|
||||
|
||||
@@ -52,6 +52,7 @@ async def get_current_metadata_user(
|
||||
):
|
||||
keycloak_sub = _keycloak_sub_from_payload(keycloak_payload)
|
||||
username = get_keycloak_preferred_username(keycloak_payload)
|
||||
email = _email_from_payload(keycloak_payload)
|
||||
try:
|
||||
user = await metadata_repo.get_user_by_keycloak_id(keycloak_sub)
|
||||
except SQLAlchemyError as exc:
|
||||
@@ -65,13 +66,14 @@ async def get_current_metadata_user(
|
||||
) from exc
|
||||
if not user or not user.is_active:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN, detail="Inactive user"
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="User authorization pending",
|
||||
)
|
||||
try:
|
||||
user = await metadata_repo.refresh_user_keycloak_snapshot(
|
||||
user,
|
||||
username=username,
|
||||
email=_email_from_payload(keycloak_payload),
|
||||
email=email,
|
||||
)
|
||||
except SQLAlchemyError as exc:
|
||||
logger.error(
|
||||
|
||||
@@ -50,6 +50,19 @@ class MetadataUserSyncResult(BaseModel):
|
||||
error: str | None = None
|
||||
|
||||
|
||||
class KeycloakUserSyncIssue(BaseModel):
|
||||
username: str
|
||||
reason: str
|
||||
|
||||
|
||||
class KeycloakUserSyncResponse(BaseModel):
|
||||
scanned: int
|
||||
created: int
|
||||
updated: int
|
||||
skipped: int
|
||||
issues: list[KeycloakUserSyncIssue] = Field(default_factory=list)
|
||||
|
||||
|
||||
class ProjectMemberCreateRequest(BaseModel):
|
||||
user_id: UUID
|
||||
project_role: ProjectRole = "viewer"
|
||||
|
||||
@@ -5,6 +5,7 @@ from uuid import UUID, uuid4
|
||||
|
||||
from cryptography.fernet import InvalidToken
|
||||
from sqlalchemy import delete, func, select
|
||||
from sqlalchemy.dialects.postgresql import insert
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.encryption import (
|
||||
@@ -158,6 +159,44 @@ class MetadataRepository:
|
||||
await self.session.refresh(user)
|
||||
return user
|
||||
|
||||
async def sync_user_snapshot_from_keycloak(
|
||||
self,
|
||||
*,
|
||||
keycloak_id: UUID,
|
||||
username: str,
|
||||
email: str,
|
||||
) -> tuple[models.User, bool]:
|
||||
user = await self.get_user_by_keycloak_id(keycloak_id)
|
||||
if user is not None:
|
||||
user.username = username
|
||||
user.email = email
|
||||
user.updated_at = _utcnow()
|
||||
await self.session.commit()
|
||||
await self.session.refresh(user)
|
||||
return user, False
|
||||
|
||||
statement = (
|
||||
insert(models.User)
|
||||
.values(
|
||||
id=uuid4(),
|
||||
keycloak_id=keycloak_id,
|
||||
username=username,
|
||||
email=email,
|
||||
role="user",
|
||||
is_active=False,
|
||||
is_superuser=False,
|
||||
)
|
||||
.on_conflict_do_nothing(index_elements=[models.User.keycloak_id])
|
||||
.returning(models.User.id)
|
||||
)
|
||||
result = await self.session.execute(statement)
|
||||
inserted_id = result.scalar_one_or_none()
|
||||
await self.session.commit()
|
||||
user = await self.get_user_by_keycloak_id(keycloak_id)
|
||||
if user is None:
|
||||
raise RuntimeError("Keycloak user synchronization did not persist a user")
|
||||
return user, inserted_id is not None
|
||||
|
||||
async def refresh_user_keycloak_snapshot(
|
||||
self,
|
||||
user: models.User,
|
||||
|
||||
@@ -0,0 +1,117 @@
|
||||
from dataclasses import dataclass
|
||||
from urllib.parse import quote
|
||||
from uuid import UUID
|
||||
|
||||
import httpx
|
||||
|
||||
|
||||
class KeycloakUserSyncConfigurationError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
class KeycloakUserSyncRequestError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class KeycloakUserSnapshot:
|
||||
keycloak_id: UUID
|
||||
username: str
|
||||
email: str | None
|
||||
service_account_client_id: str | None = None
|
||||
|
||||
|
||||
def _keycloak_admin_users_url(issuer: str) -> str:
|
||||
issuer = issuer.rstrip("/")
|
||||
marker = "/realms/"
|
||||
if not issuer or marker not in issuer:
|
||||
raise KeycloakUserSyncConfigurationError(
|
||||
"Keycloak token issuer is not a complete realm URL"
|
||||
)
|
||||
server_url, realm = issuer.rsplit(marker, 1)
|
||||
if not server_url or not realm:
|
||||
raise KeycloakUserSyncConfigurationError(
|
||||
"Keycloak token issuer is not a complete realm URL"
|
||||
)
|
||||
return f"{server_url}/admin/realms/{quote(realm, safe='')}/users"
|
||||
|
||||
|
||||
async def fetch_keycloak_users(
|
||||
*,
|
||||
issuer: str,
|
||||
bearer_token: str,
|
||||
client: httpx.AsyncClient | None = None,
|
||||
) -> list[KeycloakUserSnapshot]:
|
||||
admin_users_url = _keycloak_admin_users_url(issuer)
|
||||
if not bearer_token:
|
||||
raise KeycloakUserSyncConfigurationError(
|
||||
"Keycloak administrator access token is missing"
|
||||
)
|
||||
owns_client = client is None
|
||||
if client is None:
|
||||
client = httpx.AsyncClient(timeout=15.0)
|
||||
|
||||
try:
|
||||
try:
|
||||
snapshots: list[KeycloakUserSnapshot] = []
|
||||
first = 0
|
||||
page_size = 100
|
||||
while True:
|
||||
response = await client.get(
|
||||
admin_users_url,
|
||||
params={"first": first, "max": page_size},
|
||||
headers={
|
||||
"Accept": "application/json",
|
||||
"Authorization": f"Bearer {bearer_token}",
|
||||
},
|
||||
)
|
||||
response.raise_for_status()
|
||||
payload = response.json()
|
||||
if not isinstance(payload, list):
|
||||
raise KeycloakUserSyncRequestError(
|
||||
"Keycloak users response is not a list"
|
||||
)
|
||||
for item in payload:
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
try:
|
||||
keycloak_id = UUID(str(item.get("id", "")))
|
||||
except ValueError:
|
||||
continue
|
||||
username = str(item.get("username") or "").strip()
|
||||
if not username:
|
||||
continue
|
||||
email_value = item.get("email")
|
||||
email = str(email_value).strip() if email_value else None
|
||||
service_account_value = item.get("serviceAccountClientId")
|
||||
snapshots.append(
|
||||
KeycloakUserSnapshot(
|
||||
keycloak_id=keycloak_id,
|
||||
username=username,
|
||||
email=email,
|
||||
service_account_client_id=(
|
||||
str(service_account_value)
|
||||
if service_account_value
|
||||
else None
|
||||
),
|
||||
)
|
||||
)
|
||||
if len(payload) < page_size:
|
||||
break
|
||||
first += page_size
|
||||
return snapshots
|
||||
except httpx.HTTPStatusError as exc:
|
||||
if exc.response.status_code == 403:
|
||||
raise KeycloakUserSyncRequestError(
|
||||
"Keycloak administrator requires query-users and view-users roles"
|
||||
) from exc
|
||||
raise KeycloakUserSyncRequestError(
|
||||
"Keycloak user directory request failed"
|
||||
) from exc
|
||||
except httpx.HTTPError as exc:
|
||||
raise KeycloakUserSyncRequestError(
|
||||
"Keycloak user directory request failed"
|
||||
) from exc
|
||||
finally:
|
||||
if owns_client:
|
||||
await client.aclose()
|
||||
@@ -3,7 +3,7 @@
|
||||
"contracts": {
|
||||
"server": {
|
||||
"file": "server-v1.openapi.json",
|
||||
"sha256": "5b8ec0b469f1194e72214003a9cf427489f48de19ae31ee6492821cf64916b6d"
|
||||
"sha256": "19961572fad04f8545a4f607d4bf7276de64a84b266daecbb21d74e0975f97ef"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -20,6 +20,7 @@ from app.domain.schemas.admin_metadata import (
|
||||
ProjectMemberUpdateRequest,
|
||||
)
|
||||
from app.infra.db.metadb.repositories.metadata_repository import ProjectDbRouting
|
||||
from app.services.keycloak_user_sync import KeycloakUserSnapshot
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -176,6 +177,66 @@ async def test_batch_sync_metadata_users_returns_per_user_results(monkeypatch):
|
||||
assert admin_metadata.log_audit_event.await_count == 2
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_sync_keycloak_user_directory_preserves_authorization(monkeypatch):
|
||||
new_id = uuid4()
|
||||
existing_id = uuid4()
|
||||
repo = SimpleNamespace(
|
||||
session=SimpleNamespace(rollback=AsyncMock()),
|
||||
sync_user_snapshot_from_keycloak=AsyncMock(
|
||||
side_effect=[
|
||||
(_user(keycloak_id=new_id, is_active=False), True),
|
||||
(_user(keycloak_id=existing_id, role="admin"), False),
|
||||
]
|
||||
),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
admin_metadata,
|
||||
"fetch_keycloak_users",
|
||||
AsyncMock(
|
||||
return_value=[
|
||||
KeycloakUserSnapshot(new_id, "new-user", "new@example.com"),
|
||||
KeycloakUserSnapshot(existing_id, "admin-user", "admin@example.com"),
|
||||
KeycloakUserSnapshot(uuid4(), "no-email", None),
|
||||
KeycloakUserSnapshot(
|
||||
uuid4(),
|
||||
"service-account-client",
|
||||
"service@example.com",
|
||||
service_account_client_id="client",
|
||||
),
|
||||
]
|
||||
),
|
||||
)
|
||||
monkeypatch.setattr(admin_metadata, "log_audit_event", AsyncMock())
|
||||
|
||||
response = await admin_metadata.sync_keycloak_user_directory(
|
||||
current_user=_user(role="admin", is_superuser=True),
|
||||
keycloak_payload={"iss": "https://keycloak.example.com/realms/tjwater"},
|
||||
keycloak_access_token="admin-token",
|
||||
metadata_repo=repo,
|
||||
)
|
||||
|
||||
assert response.scanned == 4
|
||||
assert response.created == 1
|
||||
assert response.updated == 1
|
||||
assert response.skipped == 2
|
||||
assert [issue.reason for issue in response.issues] == [
|
||||
"missing_email",
|
||||
"service_account",
|
||||
]
|
||||
assert repo.sync_user_snapshot_from_keycloak.await_count == 2
|
||||
admin_metadata.fetch_keycloak_users.assert_awaited_once_with(
|
||||
issuer="https://keycloak.example.com/realms/tjwater",
|
||||
bearer_token="admin-token",
|
||||
)
|
||||
assert repo.sync_user_snapshot_from_keycloak.await_args_list[1].kwargs == {
|
||||
"keycloak_id": existing_id,
|
||||
"username": "admin-user",
|
||||
"email": "admin@example.com",
|
||||
}
|
||||
admin_metadata.log_audit_event.assert_awaited_once()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_update_metadata_user_updates_role_and_active_status(monkeypatch):
|
||||
user_id = uuid4()
|
||||
|
||||
@@ -0,0 +1,54 @@
|
||||
from uuid import uuid4
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from app.services import keycloak_user_sync
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def anyio_backend():
|
||||
return "asyncio"
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_fetch_keycloak_users_uses_admin_token_and_paginates():
|
||||
user_ids = [uuid4() for _ in range(101)]
|
||||
requested_first: list[int] = []
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
assert request.headers["authorization"] == "Bearer admin-token"
|
||||
first = int(request.url.params["first"])
|
||||
requested_first.append(first)
|
||||
page_ids = user_ids[first : first + 100]
|
||||
return httpx.Response(
|
||||
200,
|
||||
json=[
|
||||
{
|
||||
"id": str(user_id),
|
||||
"username": f"user-{first + index}",
|
||||
"email": f"user-{first + index}@example.com",
|
||||
}
|
||||
for index, user_id in enumerate(page_ids)
|
||||
],
|
||||
)
|
||||
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
||||
users = await keycloak_user_sync.fetch_keycloak_users(
|
||||
issuer="https://keycloak.example.com/realms/tjwater",
|
||||
bearer_token="admin-token",
|
||||
client=client,
|
||||
)
|
||||
|
||||
assert len(users) == 101
|
||||
assert requested_first == [0, 100]
|
||||
assert users[-1].username == "user-100"
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_fetch_keycloak_users_requires_realm_issuer():
|
||||
with pytest.raises(keycloak_user_sync.KeycloakUserSyncConfigurationError):
|
||||
await keycloak_user_sync.fetch_keycloak_users(
|
||||
issuer="",
|
||||
bearer_token="admin-token",
|
||||
)
|
||||
@@ -0,0 +1,44 @@
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
|
||||
from app.infra.db.metadb.repositories.metadata_repository import MetadataRepository
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def anyio_backend():
|
||||
return "asyncio"
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_keycloak_snapshot_sync_preserves_existing_authorization():
|
||||
keycloak_id = uuid4()
|
||||
user = SimpleNamespace(
|
||||
keycloak_id=keycloak_id,
|
||||
username="old-name",
|
||||
email="old@example.com",
|
||||
role="admin",
|
||||
is_active=True,
|
||||
is_superuser=True,
|
||||
updated_at=None,
|
||||
)
|
||||
session = SimpleNamespace(commit=AsyncMock(), refresh=AsyncMock())
|
||||
repository = MetadataRepository(session)
|
||||
repository.get_user_by_keycloak_id = AsyncMock(return_value=user)
|
||||
|
||||
result, created = await repository.sync_user_snapshot_from_keycloak(
|
||||
keycloak_id=keycloak_id,
|
||||
username="new-name",
|
||||
email="new@example.com",
|
||||
)
|
||||
|
||||
assert created is False
|
||||
assert result.username == "new-name"
|
||||
assert result.email == "new@example.com"
|
||||
assert result.role == "admin"
|
||||
assert result.is_active is True
|
||||
assert result.is_superuser is True
|
||||
session.commit.assert_awaited_once()
|
||||
session.refresh.assert_awaited_once_with(user)
|
||||
Reference in New Issue
Block a user