Fix IDOR in GET /api/users/{user_id} — email leaked to any authenticated user , Closes #25

.github/workflows/api-ci.yml

@@ -1,38 +1,51 @@
 name: API CI
-
 on:
   pull_request:
     branches: [main]
     paths:
       - "api/app/**"
       - ".github/workflows/api-ci.yml"
-
 jobs:
   lint-and-test:
     runs-on: ubuntu-latest
+    permissions:
+      contents: read
     defaults:
       run:
         working-directory: api/app
-
+    services:
+      postgres:
+        image: postgres:17-alpine
+        env:
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_DB: loomy_test
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+    env:
+      TEST_DATABASE_URL: postgresql://postgres:postgres@localhost:5432/loomy_test
     steps:
       - uses: actions/checkout@v4
-
       - name: Install uv
         uses: astral-sh/setup-uv@v4
         with:
           version: "latest"
-
       - name: Set up Python
         run: uv python install 3.12
-
       - name: Install dependencies
         run: uv sync --all-extras
-
+      - name: Run migrations (test db)
+        run: uv run alembic upgrade head
+        env:
+          DATABASE_URL: postgresql://postgres:postgres@localhost:5432/loomy_test
       - name: Ruff
         run: uv run ruff check .
-
       - name: Mypy
         run: uv run mypy .
-
       - name: Pytest
-        run: uv run pytest --tb=short -q
+        run: uv run pytest --tb=short -q

api/app/app/modules/users/presenter.py

@@ -1,7 +1,7 @@
 from app.config import settings
 from app.core.storage import is_enabled as storage_enabled, presign_get_url
 from app.modules.users.model import User
-from app.modules.users.schemas import UserResponse
+from app.modules.users.schemas import PublicUserResponse, UserResponse
 
 
 def display_name_of(user: User) -> str:
@@ -16,6 +16,16 @@ def display_name_of(user: User) -> str:
     return user.username or "Unknown"
 
 
+def public_display_name_of(user: User) -> str:
+    """Like display_name_of, but never falls back to the email local part —
+    used for PublicUserResponse where email must not be inferable."""
+    first = (user.first_name or "").strip()
+    last = (user.last_name or "").strip()
+    if first or last:
+        return f"{first} {last}".strip()
+    return user.username or "Unknown"
+
+
 def avatar_url_of(user: User) -> str | None:
     if user.avatar_key and storage_enabled():
         try:
@@ -38,3 +48,12 @@ def to_user_response(user: User) -> UserResponse:
         created_at=user.created_at,
         updated_at=user.updated_at,
     )
+
+
+def to_public_user_response(user: User) -> PublicUserResponse:
+    return PublicUserResponse(
+        id=user.id,
+        username=user.username,
+        display_name=public_display_name_of(user),
+        avatar_url=avatar_url_of(user),
+    )

api/app/app/modules/users/router.py

@@ -1,5 +1,6 @@
 import logging
 import secrets
+from typing import Union
 from uuid import UUID
 
 from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status
@@ -15,9 +16,15 @@
 )
 from app.db.session import get_db
 from app.modules.users.model import User
-from app.modules.users.presenter import to_user_response
-from app.modules.users.schemas import UserCreate, UserResponse, UserUpdate
+from app.modules.users.presenter import to_user_response, to_public_user_response
+from app.modules.users.schemas import (
+    PublicUserResponse,
+    UserCreate,
+    UserResponse,
+    UserUpdate,
+)
 from app.modules.users.service import get_user_by_id, register_user
+from app.modules.workspaces.repository import shares_workspace
 
 logger = logging.getLogger(__name__)
 
@@ -33,18 +40,20 @@
 }
 
 
-@router.get("/{user_id}", response_model=UserResponse)
+@router.get("/{user_id}", response_model=Union[UserResponse, PublicUserResponse])
 def get_user(
     user_id: UUID,
     db: Session = Depends(get_db),
     current_user: User = Depends(get_current_user),
-) -> UserResponse:
+) -> UserResponse | PublicUserResponse:
     user = get_user_by_id(db, user_id)
     if not user:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
         )
-    return to_user_response(user)
+    if shares_workspace(db, current_user.id, user_id):
+        return to_user_response(user)
+    return to_public_user_response(user)
 
 
 @router.post("", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
@@ -152,4 +161,4 @@ def delete_avatar(
     db.refresh(current_user)
     if previous_key:
         delete_object(bucket=settings.s3_bucket, key=previous_key)
-    return to_user_response(current_user)
+    return to_user_response(current_user)

api/app/app/modules/users/schemas.py

@@ -32,3 +32,16 @@ class UserResponse(BaseModel):
     updated_at: datetime
 
     model_config = {"from_attributes": True}
+
+
+class PublicUserResponse(BaseModel):
+    """Reduced profile shape returned to users who don't share a workspace
+    with the target user. Deliberately omits email / email_verified / names
+    to avoid PII leakage (see issue #25)."""
+
+    id: UUID
+    username: str
+    display_name: str
+    avatar_url: str | None = None
+
+    model_config = {"from_attributes": True}

api/app/app/modules/workspaces/repository.py

@@ -202,3 +202,23 @@ def list_workspace_invitations(
     if pending_only:
         query = query.filter(WorkspaceInvitation.accepted_at.is_(None))
     return query.order_by(WorkspaceInvitation.created_at.desc()).all()
+
+
+def shares_workspace(db: Session, user_id_a: UUID, user_id_b: UUID) -> bool:
+    """True if the two users are the same person or are both members of at
+    least one common workspace (owners are members too, via `create()`)."""
+    if user_id_a == user_id_b:
+        return True
+    member_workspace_ids = (
+        db.query(WorkspaceMember.workspace_id)
+        .filter(WorkspaceMember.user_id == user_id_b)
+    )
+    return (
+        db.query(WorkspaceMember)
+        .filter(
+            WorkspaceMember.user_id == user_id_a,
+            WorkspaceMember.workspace_id.in_(member_workspace_ids),
+        )
+        .first()
+        is not None
+    )

api/app/tests/conftest.py

@@ -1,9 +1,101 @@
+import uuid
+import os
+from collections.abc import Generator
+
 import pytest
 from fastapi.testclient import TestClient
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
 
+from app.api.deps import get_current_user
+from app.db.session import get_db
 from app.main import app
+from app.modules.users.model import User
+from app.modules.workspaces.model import Workspace, WorkspaceMember
+
+TEST_DATABASE_URL = os.environ.get(
+    "TEST_DATABASE_URL",
+    "postgresql://postgres:postgres@localhost:15432/loomy_test",
+)
+
+engine = create_engine(TEST_DATABASE_URL)
+TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+
+@pytest.fixture
+def db() -> Generator[Session, None, None]:
+    """Each test runs inside a transaction that is rolled back afterward,
+    so tests never leave residue in loomy_test and can run in any order."""
+    connection = engine.connect()
+    transaction = connection.begin()
+    session = TestingSessionLocal(bind=connection)
+
+    try:
+        yield session
+    finally:
+        session.close()
+        transaction.rollback()
+        connection.close()
 
 
 @pytest.fixture
-def client() -> TestClient:
-    return TestClient(app)
+def client(db: Session) -> Generator[TestClient, None, None]:
+    def _get_db_override() -> Generator[Session, None, None]:
+        yield db
+
+    app.dependency_overrides[get_db] = _get_db_override
+    with TestClient(app) as test_client:
+        yield test_client
+    app.dependency_overrides.clear()
+
+
+def make_user(
+    db: Session,
+    *,
+    email: str | None = None,
+    username: str | None = None,
+    first_name: str | None = "Test",
+    last_name: str | None = "User",
+) -> User:
+    """Create and persist a real User row for use in tests."""
+    unique = uuid.uuid4().hex[:8]
+    user = User(
+        email=email or f"user-{unique}@test.com",
+        username=username or f"user{unique}",
+        # Password hash is irrelevant here since tests authenticate via
+        # the get_current_user override, not a real login flow.
+        hashed_password="not-a-real-hash",
+        first_name=first_name,
+        last_name=last_name,
+        email_verified=False,
+    )
+    db.add(user)
+    db.flush()
+    db.refresh(user)
+    return user
+
+
+def make_workspace(db: Session, *, owner: User, name: str = "Test Workspace") -> Workspace:
+    """Create a workspace owned by `owner`, and add the owner as a member
+    (mirrors workspaces/repository.py's create())."""
+    unique = uuid.uuid4().hex[:8]
+    workspace = Workspace(name=name, slug=f"test-workspace-{unique}", owner_id=owner.id)
+    db.add(workspace)
+    db.flush()
+    db.refresh(workspace)
+    member = WorkspaceMember(workspace_id=workspace.id, user_id=owner.id, role="owner")
+    db.add(member)
+    db.flush()
+    return workspace
+
+
+def add_member(db: Session, *, workspace: Workspace, user: User, role: str = "member") -> None:
+    member = WorkspaceMember(workspace_id=workspace.id, user_id=user.id, role=role)
+    db.add(member)
+    db.flush()
+
+
+def auth_as(client: TestClient, user: User) -> None:
+    """Override get_current_user so requests through `client` are
+    authenticated as `user`, without going through a real JWT login."""
+    app.dependency_overrides[get_current_user] = lambda: user

api/app/tests/test_users.py

@@ -0,0 +1,89 @@
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from tests.conftest import add_member, auth_as, make_user, make_workspace
+
+
+def test_get_user_no_shared_workspace_hides_email(
+    client: TestClient, db: Session
+) -> None:
+    """Core regression test for issue #25: a user with no shared workspace
+    must not be able to retrieve another user's email via this endpoint."""
+    requester = make_user(db)
+    target = make_user(db)
+    auth_as(client, requester)
+
+    response = client.get(f"/api/users/{target.id}")
+
+    assert response.status_code == 200
+    body = response.json()
+    assert set(body.keys()) == {"id", "username", "display_name", "avatar_url"}
+    assert "email" not in body
+    assert "email_verified" not in body
+    assert "first_name" not in body
+    assert "last_name" not in body
+
+
+def test_get_user_self_lookup_returns_full_profile(
+    client: TestClient, db: Session
+) -> None:
+    user = make_user(db)
+    auth_as(client, user)
+
+    response = client.get(f"/api/users/{user.id}")
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["email"] == user.email
+    assert body["email_verified"] == user.email_verified
+
+
+def test_get_user_shared_workspace_returns_full_profile(
+    client: TestClient, db: Session
+) -> None:
+    owner = make_user(db)
+    member = make_user(db)
+    workspace = make_workspace(db, owner=owner)
+    add_member(db, workspace=workspace, user=member)
+    auth_as(client, owner)
+
+    response = client.get(f"/api/users/{member.id}")
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["email"] == member.email
+    assert body["email_verified"] == member.email_verified
+
+
+def test_get_user_display_name_never_leaks_email_local_part(
+    client: TestClient, db: Session
+) -> None:
+    """Regression test for the coderabbitai finding: when a user has no
+    first_name/last_name set, the public response's display_name must fall
+    back to username, never to the email local part."""
+    requester = make_user(db)
+    target = make_user(
+        db,
+        email="secretlocalpart@test.com",
+        first_name=None,
+        last_name=None,
+    )
+    auth_as(client, requester)
+
+    response = client.get(f"/api/users/{target.id}")
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["display_name"] == target.username
+    assert "secretlocalpart" not in body["display_name"]
+
+
+def test_get_user_not_found_returns_404(client: TestClient, db: Session) -> None:
+    import uuid
+
+    requester = make_user(db)
+    auth_as(client, requester)
+
+    response = client.get(f"/api/users/{uuid.uuid4()}")
+
+    assert response.status_code == 404