refactor(api.iam): bind tenant scope at the service dependency level #173
Conversation
…lembic - Add SQLAlchemy 2.0 with asyncpg for async database operations - Add Alembic for schema migrations - Add python-ulid for ULID support instead of UUID - Create read/write engine separation with connection pooling - Create FastAPI dependency injection for database sessions - Create SQLAlchemy declarative base with timestamp mixin - Initialize Alembic with async migration support - Create initial migration for teams table (ULID primary key) - Add comprehensive unit tests for engines and dependencies - Configure Alembic to use settings module for database URL - Enable ruff post-write hook for migration formatting Refs: AIHCM-121
- Add authzed library for SpiceDB integration - Add python-ulid for ULID support - Create ResourceType, RelationType, Permission enums (using Group not Team) - Create AuthorizationProvider protocol for swappable implementations - Implement SpiceDBClient with async methods for relationships and permissions - Create SpiceDB schema (.zed) with Tenant→Workspace→Group hierarchy - Create AuthorizationProbe for domain-oriented observability - Move ObservationContext to shared_kernel (fix architectural boundary) - Add 35 unit tests for types and probes - All 410 tests passing Refs: AIHCM-122
Resolved conflicts in authorization files by accepting remote changes: - shared_kernel/authorization/types.py (docstring fix) - shared_kernel/authorization/spicedb/client.py (_parse_reference helper)
…e dependency level
WalkthroughService constructors and dependency wiring were reworked to be tenant-scoped: UserService and GroupService now accept a TenantId ( Sequence Diagram(s)sequenceDiagram
participant Client
participant HTTP_Route as "HTTP Route"
participant GroupService as "GroupService\n(scope_to_tenant)"
participant AuthZ as "SpiceDB\n(Authorization)"
participant Repo as "GroupRepository / DB"
participant Probe as "Observability Probe"
Client->>HTTP_Route: POST /groups (name, creator_id)
HTTP_Route->>GroupService: create_group(name, creator_id)
GroupService->>AuthZ: authorize (subject, resource, scope=scope_to_tenant)
AuthZ-->>GroupService: allowed
GroupService->>Repo: persist Group(tenant_id=scope_to_tenant, ...)
Repo-->>GroupService: persisted Group
GroupService->>Probe: group_created(tenant_id=scope_to_tenant.value, ...)
GroupService-->>HTTP_Route: return Group
HTTP_Route-->>Client: 201 Created (Group)
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Suggested reviewers
🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 0
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
src/api/iam/application/services/group_service.py (1)
98-119: Docstring mentions SpiceDB, but logic is local.The tenant check is a direct
tenant_idcomparison, not SpiceDB. Update the docstring to avoid confusion.✏️ Docstring fix
- Verifies the group belongs to the scoped tenant via SpiceDB. + Verifies the group belongs to the scoped tenant via tenant_id comparison.src/api/iam/dependencies.py (1)
218-244: Changeget_group_serviceto scope to the authenticated user's tenant instead of the default tenant.
get_group_servicedepends onget_default_tenant_id, which diverges fromCurrentUser.tenant_idin multi-tenant scenarios (e.g., API-key auth wherecurrent_user.tenant_idcomes fromapi_key.tenant_id). This can create or read groups in the wrong tenant or hide valid ones. Scope services to the authenticated context by deriving the tenant fromcurrent_user:Suggested fix
+def get_current_tenant_id( + current_user: Annotated[CurrentUser, Depends(get_current_user)], +) -> TenantId: + return current_user.tenant_id + def get_group_service( group_repo: Annotated[GroupRepository, Depends(get_group_repository)], session: Annotated[AsyncSession, Depends(get_write_session)], authz: Annotated[AuthorizationProvider, Depends(get_spicedb_client)], group_service_probe: Annotated[GroupServiceProbe, Depends(get_group_service_probe)], - tenant_id: Annotated[TenantId, Depends(get_default_tenant_id)], + tenant_id: Annotated[TenantId, Depends(get_current_tenant_id)], ) -> GroupService:Apply the same change to
get_user_service.
🧹 Nitpick comments (1)
src/api/iam/application/services/user_service.py (1)
21-40: Clarify why tenant scope isn’t used yet.
scope_to_tenantis stored but not referenced inensure_user, which could imply tenant isolation that doesn’t exist. Consider documenting that users are global for now (or removing the parameter if it’s not needed yet).💡 Possible docstring tweak
- scope_to_tenant: A tenant to which this service will be scoped. + scope_to_tenant: Tenant scope reserved for future use (users are global today).
…ub.com:openshift-hyperfleet/kartograph into jsell/refactor/iam/implicit-tenant-enforcement
The API key service itself partly handles initial authentication, so it's impractical for it to require completed authentication.
Updates unit tests to match the refactored API key service that now handles authorization internally: - Add authz mock fixture for AuthorizationProvider - Update all service initialization tests to include authz parameter - Add comprehensive tests for list_api_keys authorization behavior - Fix import paths in route tests - Refactor dependency tests to separate JIT and non-JIT flows All 55 API key unit tests now passing. Co-Authored-By: Claude Sonnet 4.5 (1M context) <[email protected]>
…ub.com:openshift-hyperfleet/kartograph into jsell/refactor/iam/implicit-tenant-enforcement
There was a problem hiding this comment.
Actionable comments posted: 5
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
src/api/iam/infrastructure/api_key_repository.py (1)
216-248: Enforce required tenant_id to avoid cross-tenant listing.
tenant_idis now required in the port; allowingNonehere can return all keys if a caller passesNone. Enforce non-null and align the docstring.Proposed fix
@@ - api_key_ids: Optional list of specific API key IDs to include - tenant_id: Optional tenant to scope the list to + tenant_id: Required tenant to scope the list to + api_key_ids: Optional list of specific API key IDs to include @@ - if tenant_id is not None: - conditions.append(APIKeyModel.tenant_id == tenant_id.value) + if tenant_id is None: + raise ValueError("tenant_id is required") + conditions.append(APIKeyModel.tenant_id == tenant_id.value)src/api/iam/application/services/api_key_service.py (1)
43-57: Docstring still mentionsscope_to_tenant.
Line 55 references a parameter that no longer exists; it should describe theauthzdependency.✏️ Suggested doc fix
- scope_to_tenant: A tenant to which this service will be scoped. + authz: Authorization provider (SpiceDB client)
🤖 Fix all issues with AI agents
In `@src/api/iam/application/observability/api_key_service_probe.py`:
- Around line 61-63: The Protocol method stub
api_key_list_retrieval_failed(user_id: str, reason: str) -> None is missing the
ellipsis used by the other Protocol methods; update the method body to contain a
single ellipsis (...) so it follows the class's Protocol pattern (i.e., replace
the empty docstring-only body with ... in api_key_list_retrieval_failed).
In `@src/api/iam/dependencies/authentication.py`:
- Around line 14-36: The module currently creates oauth2_scheme at import via
_create_oauth2_scheme() which will raise and fail import if get_oidc_settings()
validation ever errors; change this by wrapping the creation in a try/except
(catch broad Exception), log a warning/error, and set oauth2_scheme to None (or
an explicitly disabled placeholder) so import never crashes; update the
oauth2_scheme variable's typing to Optional[OAuth2AuthorizationCodeBearer] and
ensure callers handle None (or provide a get_oauth2_scheme() accessor that calls
_create_oauth2_scheme() inside the same try/except from main.py's safe context);
also reconcile the docstring about client_secret being "required" with the
OIDCSettings default (update docstring or make the field required) so
documentation matches code.
In `@src/api/iam/dependencies/user.py`:
- Around line 140-155: The get_user_service docstring incorrectly documents a
non-existent tenant_id parameter; update the Args section to remove tenant_id
and describe current_user (Annotated[CurrentUser,
Depends(get_current_user_no_jit)]) instead, explaining that current_user
provides the authenticated user's context for scoping the service; ensure the
description fits the existing parameters and references the get_user_service
function and the current_user parameter name.
- Around line 167-184: The docstring for get_current_user is out of date; remove
or update the "Args" section to reflect the actual parameters (current_user:
CurrentUser from get_current_user_no_jit, user_service: UserService from
get_user_service, and token: str | None from oauth2_scheme) and drop references
to x_api_key, validator, api_key_service, and auth_probe; ensure the top-line
description still mentions JWT/API Key behavior if accurate and list only the
real params (CurrentUser, get_current_user_no_jit, get_user_service,
oauth2_scheme) so the docstring matches the function signature.
In `@src/api/iam/presentation/routes.py`:
- Around line 456-463: The code constructs UserId(value=user_id) without
validation in the route before calling service.list_api_keys; replace this with
validated parsing using UserId.from_string(user_id) (or equivalent) and if it
raises ValueError catch it and return a 400 error response (or raise the same
HTTPBadRequest used elsewhere) so empty/whitespace user_id is rejected; update
the block that prepares created_by_user_id for service.list_api_keys to validate
user_id via UserId.from_string and only pass a UserId on success, otherwise
respond 400.
🧹 Nitpick comments (2)
src/api/iam/presentation/routes.py (1)
324-353: Consider tracking the TODO for admin-only tenant deletion.
If you want, I can draft the admin guard (role check/dependency) or open a follow-up issue.src/api/iam/application/services/api_key_service.py (1)
63-81: Docstring should mentiontenant_id.
The new parameter is part of the public API; please document it for callers.✏️ Suggested doc fix
Args: created_by_user_id: The user who is creating this key (audit trail) name: A descriptive name for the key expires_in_days: Number of days until expiration + tenant_id: Tenant that owns the API key
| def _create_oauth2_scheme() -> OAuth2AuthorizationCodeBearer: | ||
| """Create OAuth2 security scheme for Swagger UI integration. | ||
|
|
||
| Uses the OIDC issuer URL to configure authorization code flow endpoints. | ||
| This enables Swagger UI's Authorize button to work with Keycloak. | ||
| """ | ||
| issuer = get_oidc_settings().issuer_url | ||
|
|
||
| return OAuth2AuthorizationCodeBearer( | ||
| authorizationUrl=f"{issuer}/protocol/openid-connect/auth", | ||
| tokenUrl=f"{issuer}/protocol/openid-connect/token", | ||
| refreshUrl=f"{issuer}/protocol/openid-connect/token", | ||
| scopes={ | ||
| "openid": "OpenID Connect", | ||
| "profile": "User profile", | ||
| "email": "User email", | ||
| }, | ||
| auto_error=False, | ||
| ) | ||
|
|
||
|
|
||
| # Create OAuth2 security scheme for Swagger UI integration | ||
| oauth2_scheme = _create_oauth2_scheme() |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
# First, examine the settings file to understand OIDC configuration
rg -n "class OIDCSettings|issuer_url|client_id|client_secret|swagger_client_id" src/api/infrastructure/settings.pyRepository: openshift-hyperfleet/kartograph
Length of output: 600
🏁 Script executed:
# Look for the get_oidc_settings() implementation and error handling
rg -n "def get_oidc_settings|class.*Settings" src/api/infrastructure/settings.py -A 10Repository: openshift-hyperfleet/kartograph
Length of output: 4115
🏁 Script executed:
# Check the authentication.py file structure and any error handling
cat -n src/api/iam/dependencies/authentication.pyRepository: openshift-hyperfleet/kartograph
Length of output: 2575
🏁 Script executed:
# Search for how oauth2_scheme is imported and used elsewhere
rg -n "oauth2_scheme|from.*authentication import" src/api --type pyRepository: openshift-hyperfleet/kartograph
Length of output: 750
🏁 Script executed:
# Look for app startup or initialization code that might handle OIDC errors
rg -n "OIDCSettings|oidc_settings|get_oidc_settings" src/api -B 2 -A 2 --type py | head -50Repository: openshift-hyperfleet/kartograph
Length of output: 3085
🏁 Script executed:
# Check exact field definitions in OIDCSettings with full context
sed -n '298,360p' src/api/infrastructure/settings.pyRepository: openshift-hyperfleet/kartograph
Length of output: 2307
🏁 Script executed:
# Look at main.py to understand error handling for OIDC
rg -n "configure_swagger_oauth2|get_oidc_settings" src/api/main.py -B 3 -A 8Repository: openshift-hyperfleet/kartograph
Length of output: 2064
🏁 Script executed:
# Check if authentication.py import happens conditionally or always
rg -n "from.*authentication|import.*oauth2_scheme" src/api/main.pyRepository: openshift-hyperfleet/kartograph
Length of output: 57
🏁 Script executed:
# Look for any try/except wrapping the module-level code
head -50 src/api/main.pyRepository: openshift-hyperfleet/kartograph
Length of output: 1737
Architectural concern: Import-time OAuth2 scheme creation lacks error handling, though practical risk is low.
oauth2_scheme is created at module import time (line 36) from get_oidc_settings(). While all OIDC configuration fields have defaults (including client_secret: SecretStr = Field(default=SecretStr(""), ...)), this means import-time failure is unlikely in current practice. However, if OIDCSettings validation ever fails, the module import will fail without error handling, unlike main.py which wraps identical calls in try/except (lines 233-238, 258-262).
Consider either:
- Adding try/except around
_create_oauth2_scheme()to gracefully handle missing/invalid OIDC settings - Deferring scheme creation to a function called from
main.py's error-safe context, or - Documenting that all OIDC fields have safe defaults and import-time failure is expected to be impossible
Note: The docstring states client_secret is "required," but the field definition has default=SecretStr(""), creating a documentation/code mismatch.
🤖 Prompt for AI Agents
In `@src/api/iam/dependencies/authentication.py` around lines 14 - 36, The module
currently creates oauth2_scheme at import via _create_oauth2_scheme() which will
raise and fail import if get_oidc_settings() validation ever errors; change this
by wrapping the creation in a try/except (catch broad Exception), log a
warning/error, and set oauth2_scheme to None (or an explicitly disabled
placeholder) so import never crashes; update the oauth2_scheme variable's typing
to Optional[OAuth2AuthorizationCodeBearer] and ensure callers handle None (or
provide a get_oauth2_scheme() accessor that calls _create_oauth2_scheme() inside
the same try/except from main.py's safe context); also reconcile the docstring
about client_secret being "required" with the OIDCSettings default (update
docstring or make the field required) so documentation matches code.
Improves the list_api_keys endpoint to properly validate input and document responses using FastAPI conventions: - Add proper validation for user_id query parameter - Return 400 for empty/whitespace user_id values - Add explicit FastAPI response documentation (200, 400, 500) - Fix exception handling to re-raise HTTPException properly - Add test for empty user_id validation Routes now properly document all possible response codes and their meanings in the OpenAPI schema. Co-Authored-By: Claude Sonnet 4.5 (1M context) <[email protected]>
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
src/api/iam/infrastructure/api_key_repository.py (1)
236-249: Validatetenant_idbefore returning early.When
api_key_idsis empty, the method returns[]before enforcingtenant_idbeing required, which can mask missing-tenant_id bugs. Move the validation ahead of the early return for consistent behavior.🔧 Proposed fix
- if api_key_ids is not None: - if not api_key_ids: - # Empty list means no results - return [] - # Extract string values from APIKeyId objects - id_values = [id_val.value for id_val in api_key_ids] - conditions.append(APIKeyModel.id.in_(id_values)) - if tenant_id is None: raise ValueError("tenant_id is required") conditions.append(APIKeyModel.tenant_id == tenant_id.value) + + if api_key_ids is not None: + if not api_key_ids: + # Empty list means no results + return [] + # Extract string values from APIKeyId objects + id_values = [id_val.value for id_val in api_key_ids] + conditions.append(APIKeyModel.id.in_(id_values))src/api/iam/application/services/api_key_service.py (1)
43-60: Docstring missingauthzdependency.
__init__now requiresauthz, but the Args section doesn’t mention it.✏️ Doc fix
Args: session: Database session for transaction management api_key_repository: Repository for API key persistence + authz: Authorization provider (SpiceDB client) probe: Optional domain probe for observabilityAs per coding guidelines, Document agent capabilities and behavior in docstrings.
🤖 Fix all issues with AI agents
In `@src/api/iam/application/services/api_key_service.py`:
- Around line 138-165: The code forces a user filter by default (search_by_user
defaults to current_user), preventing tenant admins from seeing all keys; change
the logic so search_by_user is only set when created_by_user_id is provided,
call self._api_key_repository.list with created_by_user_id only when
created_by_user_id is not None (otherwise omit that argument), and update the
probe calls (self._probe.api_key_list_retrieved / api_key_list_retrieval_failed)
to log user_id as created_by_user_id.value when provided else fallback to
current_user.user_id.value; keep the SpiceDB lookup via
self._authz.lookup_resources and the APIKeyId wrapping unchanged.
In `@src/api/iam/presentation/routes.py`:
- Around line 466-476: The current check uses "if user_id:" which skips
validation for empty strings; change the condition to "if user_id is not None"
so that UserId.from_string(value=user_id) runs for empty or whitespace values
and will raise the intended ValueError; ensure the except block still raises the
HTTPException with status.HTTP_400_BAD_REQUEST and the same detail message
(references: user_id variable, filter_user_id assignment, and
UserId.from_string).
🧹 Nitpick comments (1)
src/api/iam/presentation/routes.py (1)
316-330: Access control TODO for tenant deletion is still open.Do you want me to draft a dependency/guard (e.g., admin check) and open an issue for tracking?
| # Use SpiceDB to find all API keys the current user can view | ||
| search_by_user: UserId = ( | ||
| created_by_user_id if created_by_user_id else current_user.user_id | ||
| ) | ||
| try: | ||
| api_key_ids = await self._authz.lookup_resources( | ||
| resource_type=ResourceType.API_KEY.value, | ||
| permission=Permission.VIEW.value, | ||
| subject=format_subject(ResourceType.USER, current_user.user_id.value), | ||
| ) | ||
|
|
||
| self._probe.api_key_list_retrieved( | ||
| user_id=created_by_user_id.value if created_by_user_id else "filtered", | ||
| count=len(keys), | ||
| ) | ||
| return keys | ||
| keys = await self._api_key_repository.list( | ||
| api_key_ids=[APIKeyId(value=key) for key in api_key_ids], | ||
| tenant_id=tenant_id, | ||
| created_by_user_id=search_by_user, | ||
| ) | ||
|
|
||
| self._probe.api_key_list_retrieved( | ||
| user_id=search_by_user.value, | ||
| count=len(keys), | ||
| ) | ||
| return keys | ||
| except Exception as e: | ||
| self._probe.api_key_list_retrieval_failed( | ||
| user_id=search_by_user.value, | ||
| reason=repr(e), | ||
| ) | ||
| raise |
There was a problem hiding this comment.
Don’t force a user filter when created_by_user_id is omitted.
When created_by_user_id is None, you still filter by current_user, which prevents tenant admins from listing all keys they’re authorized to view. Let SpiceDB determine visibility and only apply the user filter when explicitly requested.
🐛 Proposed fix
- search_by_user: UserId = (
- created_by_user_id if created_by_user_id else current_user.user_id
- )
+ search_by_user: UserId | None = created_by_user_id
+ probe_user = created_by_user_id or current_user.user_id
try:
api_key_ids = await self._authz.lookup_resources(
resource_type=ResourceType.API_KEY.value,
permission=Permission.VIEW.value,
subject=format_subject(ResourceType.USER, current_user.user_id.value),
)
keys = await self._api_key_repository.list(
api_key_ids=[APIKeyId(value=key) for key in api_key_ids],
tenant_id=tenant_id,
created_by_user_id=search_by_user,
)
self._probe.api_key_list_retrieved(
- user_id=search_by_user.value,
+ user_id=probe_user.value,
count=len(keys),
)
return keys
except Exception as e:
self._probe.api_key_list_retrieval_failed(
- user_id=search_by_user.value,
+ user_id=probe_user.value,
reason=repr(e),
)
raise🤖 Prompt for AI Agents
In `@src/api/iam/application/services/api_key_service.py` around lines 138 - 165,
The code forces a user filter by default (search_by_user defaults to
current_user), preventing tenant admins from seeing all keys; change the logic
so search_by_user is only set when created_by_user_id is provided, call
self._api_key_repository.list with created_by_user_id only when
created_by_user_id is not None (otherwise omit that argument), and update the
probe calls (self._probe.api_key_list_retrieved / api_key_list_retrieval_failed)
to log user_id as created_by_user_id.value when provided else fallback to
current_user.user_id.value; keep the SpiceDB lookup via
self._authz.lookup_resources and the APIKeyId wrapping unchanged.
Changes the probe's user_id parameter to represent the filter being applied (which can be None) rather than the user performing the operation: - Update probe protocol and implementation to accept user_id: str | None - Rename logged field from user_id to filter_user_id for clarity - Fix service to pass created_by_user_id.value instead of repr() - Update all tests to expect None when no filter is specified This makes the probe semantics more accurate - it now logs the created_by_user_id filter parameter, which can be None to indicate "show all viewable keys" rather than "show only keys created by this specific user". Co-Authored-By: Claude Sonnet 4.5 (1M context) <[email protected]>
1 participant
Summary
To best enforce tenant isolation, and improve the developer experience, tenancy is now bound to the service at the dependency level.
Concretely this means that, for example, a single instance of the user or group service cannot be used to retrieve data from two different tenants. Further, by binding the tenant to the service at the dependency injection level, downstream method signatures are simplified as the tenant scope is implicit.
Summary by CodeRabbit
Refactor
Improvements
Tests
✏️ Tip: You can customize this high-level summary in your review settings.