diff --git a/.claude/CLAUDE.md b/.claude/CLAUDE.md index 1f0f8e7d..9dd1023d 100644 --- a/.claude/CLAUDE.md +++ b/.claude/CLAUDE.md @@ -180,11 +180,17 @@ class CreateGroup: ... ``` -Plugin hooks fire from inside `execute()`. Side-effecting Okta/notification work is spawned as +Plugin hooks fire from inside `execute()`. The plugin interface is **async** (see the Plugin +system section): each hook call returns coroutines that Access awaits via `run_hooks_to_completion` +in `api/plugins/_async_dispatch.py`, which uses `asyncio.wait` (not `gather`, so one hook failing +or a cancelled request can't tear down its siblings' in-flight I/O) and logs a failing hook at +**ERROR** — a plugin's failure is surfaced, not swallowed, yet never breaks the committed +operation; plugins are expected to catch their own expected/noisy errors. Separately, +side-effecting Okta/notification work is spawned as `asyncio` tasks after the local DB state commits, then drained at the end of `execute()` via `drain_fan_out_tasks` in `api/operations/_fan_out.py` — which surfaces task failures (logged at -WARNING) instead of dropping them, and uses `asyncio.wait` so a cancelled request doesn't tear -down in-flight Okta calls. These fan-out tasks do network I/O only, never `db.session` access. Operations fall into two broad groups — request +WARNING) instead of dropping them, and likewise uses `asyncio.wait` so a cancelled request doesn't +tear down in-flight Okta calls. These fan-out tasks do network I/O only, never `db.session` access. Operations fall into two broad groups — request operations (create/approve/reject for access, group, and role requests) and resource operations (create/delete/modify for apps, groups, tags, users, and roles). The exact set changes as features land, so read `api/operations/` for the current, authoritative list — one class per @@ -497,6 +503,19 @@ implementations live in that operator's own private repo. Each operator writes t implementations (for notifications, conditional access, etc.); operator-specific behavior belongs in one of those plugins, not upstream. +**The plugin interface is async (Access 2.0).** `pluggy` never awaits, so a hook call returns a +list of coroutines that Access drives to completion via `run_hooks_to_completion` +(`api/plugins/_async_dispatch.py`). Consequently hook implementations across all four surfaces — +notifications, conditional access, `metrics_reporter` (`record_*` etc.; `batch_metrics` is an +async context manager), and the app-group-lifecycle *lifecycle* hooks — must be `async def`, and +`verify_async_impls` fails fast at hook load if one is registered as a plain `def`. Do blocking +I/O off the event loop: prefer a native async client (`httpx`, an SDK's async client) and fall +back to `await asyncio.to_thread(...)` only for sync-only dependencies. App-group-lifecycle hooks +that mutate state receive an `AsyncSession` (await ORM calls on it; `session.add(...)` is sync); +the pure metadata/config/status/validation hooks stay **sync**. Plugin-contributed +`access.commands` CLI commands run as ordinary Click commands and must drive their own +`asyncio.run(...)`. The README's plugin section documents this in full. + For hook signatures, adding a new plugin type, and implementing an operator override, read the hookspecs in `api/plugins/`. diff --git a/POST_MIGRATION_TODO.md b/POST_MIGRATION_TODO.md index d5b72b88..516b12a2 100644 --- a/POST_MIGRATION_TODO.md +++ b/POST_MIGRATION_TODO.md @@ -124,71 +124,6 @@ clean `--snapshot-update` ergonomics. --- -## Plugin Interface Modernization - -### 18. Make plugin interfaces async; deprecate the sync hooks - -The four plugin types (`notifications`, `conditional_access`, -`app_group_lifecycle`, `metrics_reporter`) currently expose synchronous -hooks via `pluggy`. The application is now async, so every call site -bridges the sync hook — and the bridge differs by call site, which is -exactly the smell that motivates going natively async: - -- `app_group_lifecycle` hooks (which receive `session=`) run through - `AsyncSession.run_sync`, which hands them a working sync `Session` on - the greenlet bridge (`create_group`, `delete_group`, `modify_group_*`). -- The syncer's `access_expiring_*` `notifications` hooks also run through - `db.session.run_sync` — the syncer is a batch CLI job holding - AsyncSession-bound ORM rows, so run_sync keeps those rows on the - session's own thread instead of handing them to an `anyio` worker - thread (which was the original cross-thread hazard). Every ORM - relationship is `lazy="raise_on_sql"`, so a hook can only read what the - syncer eager-loaded regardless — extend the `joinedload`s there if a - notification plugin needs a wider graph. -- `notifications` / `conditional_access` hooks fired from request-path - operations still run **inline on the event loop**, so a plugin doing - slow network I/O blocks the request. Moving that slow tail off the hot - path is item 10 (`BackgroundTasks`); `conditional_access` is harder - because its return value gates the request. - -Making the hooks natively async collapses all three bridges into one -`await`. Plugins should eventually follow the app. - -Strategy mirrors the existing `DeprecationWarning` pattern in -`api/plugins/notifications.py:48-74`: - -1. Duplicate every hook spec to add an `_async` variant — - `access_request_created_async`, `notify_user_async`, etc. -2. Keep both for one major version. Sync hooks emit - `DeprecationWarning` when called. -3. Remove the sync set at the next major bump. - -Plugins authored in this window can pick either flavor; the application -prefers the async hook when both are registered. - -Rules to document with the async hook specs: -- Hooks must not import `api.extensions.db` — they receive everything - they need (the lifecycle hooks' `session` argument is a sync - `Session` shim today and becomes an `AsyncSession` for the `_async` - variants). -- **Plugin-contributed CLI commands** (the `access.commands` entry - point) now run inside the CLI's `asyncio.run` boundary; commands - written against the old sync `db.session` will break. They should - declare `async def` bodies and await session calls — document this - in the plugin guide alongside the async hook rollout. - -### 19. Pre-2.0 release checklist: drop deprecated plugin parameters - -Several `access_expiring_*` hooks already carry `DeprecationWarning` -for legacy parameters: `groups`, `roles`, `users` on -`access_expiring_owner`; `groups` on `access_expiring_user`. Their -deprecation window ends at 2.0 — the parameters and their -backward-compat shims must be removed before tagging the release. - -`api/plugins/notifications.py:48-74` is the canonical list. - ---- - ## Security follow-ups (out of scope for the migration PR) ### 20. Nonce-based CSP diff --git a/README.md b/README.md index b8027602..3a07a8af 100644 --- a/README.md +++ b/README.md @@ -471,6 +471,17 @@ Plugins in Access follow the conventions defined by the [Python pluggy framework Plugins can also extend the `access` CLI with new commands via setuptools entry points — see [health_check_plugin](https://github.com/discord/access/tree/main/examples/plugins/health_check_plugin), which adds an `access health` command. +### Async hook interface (Access 2.0) + +As of Access 2.0 the plugin hooks are **async** — Access is an async application and awaits each hook directly on the event loop, with no thread/greenlet bridge. When writing a plugin: + +- **Declare hook implementations with `async def`.** This applies to the notification hooks, the conditional-access hooks, the metrics `record_*` / `set_global_tags` / `flush` hooks, and the app-group-lifecycle *lifecycle* hooks (`group_created`, `group_updated`, `group_deleted`, `group_members_added`, `group_members_removed`, `sync_all_group_membership`). Access fails fast at startup with a clear error if one of these is registered as a plain `def`. +- **Do blocking I/O off the event loop.** Prefer a native async client — `httpx`, `aiohttp`, or an SDK's async client (e.g. Slack's `AsyncWebClient`, as in [notifications_slack](https://github.com/discord/access/tree/main/examples/plugins/notifications_slack)). Only when a dependency is synchronous-only (no async client available) should you wrap its blocking calls in `await asyncio.to_thread(...)`. Never run blocking network calls directly in an `async def` hook. +- **Lifecycle hooks receive an `AsyncSession`.** Await ORM calls on it (`await session.scalars(...)`); `session.add(...)` is synchronous. Never import `api.extensions.db` — everything a hook needs is passed in. +- **Metrics `batch_metrics` is an async context manager** — use it as `async with metrics.batch_metrics(): await metrics.record_counter(...)`. +- **These hooks stay synchronous:** the app-group-lifecycle metadata/config-schema/status-schema/validation hooks (`get_plugin_metadata`, `get_plugin_*_config_properties`, `validate_plugin_*_config`, `get_plugin_*_status_properties`). They are pure schema accessors with no I/O. +- **Plugin-contributed CLI commands** (the `access.commands` entry point) run as ordinary Click commands, so they must drive their own async boundary — write an `async def` body and run it with `asyncio.run(...)`, as [health_check_plugin](https://github.com/discord/access/tree/main/examples/plugins/health_check_plugin) does. + ### Installing a Plugin in the Docker Container Below is an example Dockerfile that would install the example notification plugin into the Access Docker container, which was built above using the top-level application [Dockerfile](https://github.com/discord/access/blob/main/Dockerfile). The plugin is copied into the `/app/plugins` directory and installed with `uv` into the application virtualenv at `/app/.venv` (the environment gunicorn and the `access` CLI run from). diff --git a/api/manage.py b/api/manage.py index 785956d2..0451962c 100644 --- a/api/manage.py +++ b/api/manage.py @@ -338,6 +338,7 @@ async def sync_app_group_memberships() -> None: """Invoke the periodic membership sync hook for all apps with app group lifecycle plugins configured.""" from api.extensions import db from api.models import App + from api.plugins._async_dispatch import run_hooks_to_completion from api.plugins.app_group_lifecycle import get_app_group_lifecycle_hook click.echo("Starting app group lifecycle plugin sync") @@ -360,14 +361,18 @@ async def sync_app_group_memberships() -> None: for app in apps: click.echo(f"Syncing app '{app.name}' (plugin: {app.app_group_lifecycle_plugin})") + # App-group-lifecycle hooks are native async: awaited directly + # with the AsyncSession, no run_sync bridge. run_hooks_to_completion uses + # asyncio.wait (not gather) and logs any plugin failure itself. + _, exceptions = await run_hooks_to_completion( + hook.sync_all_group_membership(session=db.session, app=app, plugin_id=app.app_group_lifecycle_plugin), + context=f"sync_all_group_membership for app '{app.name}'", + ) + if exceptions: + await db.session.rollback() + click.echo(f" ✗ Failed to sync app '{app.name}': {exceptions[0]}", err=True) + continue try: - # App-group-lifecycle hooks are sync plugin code that expects a - # sync Session; run them through the session's greenlet bridge. - await db.session.run_sync( - lambda session, app=app: hook.sync_all_group_membership( # type: ignore[misc] - session=session, app=app, plugin_id=app.app_group_lifecycle_plugin - ) - ) await db.session.commit() click.echo(f" ✓ Synced app '{app.name}'") except Exception as e: diff --git a/api/middleware.py b/api/middleware.py index ac1ee329..8aea99c4 100644 --- a/api/middleware.py +++ b/api/middleware.py @@ -32,6 +32,7 @@ from api.config import settings from api.context import RequestContext, reset_request_context, set_request_context from api.extensions import _session_scope, db +from api.plugins._async_dispatch import run_hooks_to_completion from api.plugins.metrics_reporter import get_metrics_reporter_hook CSP = ( @@ -187,9 +188,9 @@ async def _send(message: Message) -> None: await self.app(scope, receive, _send) finally: duration_ms = (time.perf_counter() - start) * 1000.0 - self._record(scope.get("method", ""), status_code, duration_ms) + await self._record(scope.get("method", ""), status_code, duration_ms) - def _record(self, method: str, status_code: int, duration_ms: float) -> None: + async def _record(self, method: str, status_code: int, duration_ms: float) -> None: try: hook = get_metrics_reporter_hook() except Exception: @@ -197,7 +198,15 @@ def _record(self, method: str, status_code: int, duration_ms: float) -> None: return tags = {"method": method, "status": str(status_code)} try: - hook.record_counter(metric_name="requests", value=1, tags=tags) - hook.record_histogram(metric_name="request.duration", value=duration_ms, tags={**tags, "unit": "ms"}) + # run_hooks_to_completion uses asyncio.wait (not gather) and logs any + # per-plugin failure itself; emit is best-effort and never fails the request. + await run_hooks_to_completion( + hook.record_counter(metric_name="requests", value=1, tags=tags), + context="metrics requests counter", + ) + await run_hooks_to_completion( + hook.record_histogram(metric_name="request.duration", value=duration_ms, tags={**tags, "unit": "ms"}), + context="metrics request.duration histogram", + ) except Exception: self._logger.exception("metrics_reporter emit failed; continuing") diff --git a/api/operations/approve_access_request.py b/api/operations/approve_access_request.py index b3410c30..b536ef8c 100644 --- a/api/operations/approve_access_request.py +++ b/api/operations/approve_access_request.py @@ -13,7 +13,6 @@ from api.models import AccessRequest, AccessRequestStatus, AppGroup, OktaGroup, OktaUser, RoleGroup from api.operations.constraints import CheckForReason from api.operations.modify_group_users import ModifyGroupUsers -from api.plugins import get_notification_hook from api.schemas import AuditLogSchema, EventType @@ -38,8 +37,6 @@ def __init__( self.notify = notify - self.notification_hook = get_notification_hook() - async def execute(self) -> AccessRequest: # Lock the request row for the duration of the transaction so two # concurrent approvers can't both pass the pending-state guard below @@ -147,34 +144,4 @@ async def execute(self) -> AccessRequest: notify=self.notify, ).execute() - # Now handled inside ModifyGroupUsers - # self.access_request.approved_membership_id = ( - # db.session.query(OktaUserGroupMember).filter( - # OktaUserGroupMember.user_id == self.access_request.requester_user_id - # ) - # .filter( - # OktaUserGroupMember.group_id == self.access_request.requested_group_id - # ) - # .filter(OktaUserGroupMember.role_group_map_id.is_(None)) - # .filter( - # OktaUserGroupMember.is_owner == self.access_request.request_ownership - # ) - # .order_by(OktaUserGroupMember.created_at.desc()) - # .first() - # .id - # ) - # db.session.commit() - - # requester = db.session.get(OktaUser, self.access_request.requester_user_id) - - # approvers = get_all_possible_request_approvers(self.access_request) - - # self.notification_hook.access_request_completed( - # access_request=self.access_request, - # group=group, - # requester=requester, - # approvers=approvers, - # notify_requester=True, - # ) - return access_request diff --git a/api/operations/approve_group_request.py b/api/operations/approve_group_request.py index 0cb9fa31..a2e8adff 100644 --- a/api/operations/approve_group_request.py +++ b/api/operations/approve_group_request.py @@ -25,7 +25,7 @@ from api.operations.constraints.check_for_self_add import CheckForSelfAdd from api.operations.create_group import CreateGroup from api.operations.modify_group_users import ModifyGroupUsers -from api.plugins import get_notification_hook +from api.plugins import NotificationHook, send_notification from api.schemas import AuditLogSchema, EventType @@ -47,7 +47,6 @@ def __init__( self.approval_reason = approval_reason self.notify = notify self.bypass_self_approval = bypass_self_approval - self.notification_hook = get_notification_hook() async def execute(self) -> Optional[GroupRequest]: # Lock the request row for the transaction so concurrent approvers @@ -278,7 +277,8 @@ async def execute(self) -> Optional[GroupRequest]: if self.notify: requester = await db.session.get(OktaUser, group_request.requester_user_id) approvers = await get_all_possible_request_approvers(group_request) - self.notification_hook.access_group_request_completed( + await send_notification( + NotificationHook.ACCESS_GROUP_REQUEST_COMPLETED, group_request=group_request, group=created_group, requester=requester, diff --git a/api/operations/approve_role_request.py b/api/operations/approve_role_request.py index cb2c0c8c..0535479c 100644 --- a/api/operations/approve_role_request.py +++ b/api/operations/approve_role_request.py @@ -13,7 +13,6 @@ from api.models import AccessRequestStatus, AppGroup, OktaGroup, OktaUser, RoleGroup, RoleRequest from api.operations.constraints import CheckForReason from api.operations.modify_role_groups import ModifyRoleGroups -from api.plugins import get_notification_hook from api.schemas import AuditLogSchema, EventType @@ -38,8 +37,6 @@ def __init__( self.notify = notify - self.notification_hook = get_notification_hook() - async def execute(self) -> RoleRequest: # Lock the request row for the transaction so concurrent approvers # can't both pass the pending-state guard and double-grant. `of=` keeps diff --git a/api/operations/create_access_request.py b/api/operations/create_access_request.py index fd458a4e..ab8eb589 100644 --- a/api/operations/create_access_request.py +++ b/api/operations/create_access_request.py @@ -23,7 +23,7 @@ from api.models.okta_group import get_group_managers from api.operations.approve_access_request import ApproveAccessRequest from api.operations.reject_access_request import RejectAccessRequest -from api.plugins import get_conditional_access_hook, get_notification_hook +from api.plugins import ConditionalAccessHook, NotificationHook, evaluate_conditional_access, send_notification from api.schemas import AuditLogSchema, EventType @@ -48,9 +48,6 @@ def __init__( self.request_approvers = select() - self.conditional_access_hook = get_conditional_access_hook() - self.notification_hook = get_notification_hook() - async def execute(self) -> Optional[AccessRequest]: requester = await db.session.get(OktaUser, self.requester_user_id) @@ -131,7 +128,8 @@ async def execute(self) -> Optional[AccessRequest]: ) ) - conditional_access_responses = self.conditional_access_hook.access_request_created( + conditional_access_responses = await evaluate_conditional_access( + ConditionalAccessHook.ACCESS_REQUEST_CREATED, access_request=access_request, group=group, group_tags=[active_tag_map.enabled_active_tag for active_tag_map in group.active_group_tags], @@ -156,7 +154,8 @@ async def execute(self) -> Optional[AccessRequest]: return access_request - self.notification_hook.access_request_created( + await send_notification( + NotificationHook.ACCESS_REQUEST_CREATED, access_request=access_request, group=group, requester=requester, diff --git a/api/operations/create_group.py b/api/operations/create_group.py index 724978fd..d56a3f67 100644 --- a/api/operations/create_group.py +++ b/api/operations/create_group.py @@ -8,7 +8,7 @@ from api.extensions import db from api.models import App, AppGroup, AppTagMap, OktaGroup, OktaGroupTagMap, OktaUser, RoleGroup, Tag -from api.plugins.app_group_lifecycle import invoke_app_group_lifecycle_hook +from api.plugins.app_group_lifecycle import AppGroupLifecycleHook, invoke_app_group_lifecycle_hook from api.services import okta from api.schemas import AuditLogSchema, EventType @@ -120,7 +120,7 @@ async def execute(self, *, _group: Optional[T] = None) -> T: await db.session.commit() # Invoke app group lifecycle plugin hook, if configured - await invoke_app_group_lifecycle_hook("group_created", group=self.group) + await invoke_app_group_lifecycle_hook(AppGroupLifecycleHook.GROUP_CREATED, group=self.group) # Audit logging email = None diff --git a/api/operations/create_group_request.py b/api/operations/create_group_request.py index bafcfc05..72582025 100644 --- a/api/operations/create_group_request.py +++ b/api/operations/create_group_request.py @@ -22,7 +22,7 @@ from api.models.tag import coalesce_ended_at from api.operations.approve_group_request import ApproveGroupRequest from api.operations.reject_group_request import RejectGroupRequest -from api.plugins import get_conditional_access_hook, get_notification_hook +from api.plugins import ConditionalAccessHook, NotificationHook, evaluate_conditional_access, send_notification from api.schemas import AuditLogSchema, EventType @@ -51,9 +51,6 @@ def __init__( self.requested_ownership_ending_at = requested_ownership_ending_at self.request_reason = request_reason - self.conditional_access_hook = get_conditional_access_hook() - self.notification_hook = get_notification_hook() - async def execute(self) -> Optional[GroupRequest]: requester = await db.session.get(OktaUser, self.requester_user_id) @@ -183,7 +180,8 @@ async def execute(self) -> Optional[GroupRequest]: ) # Check conditional access hook - conditional_access_responses = self.conditional_access_hook.group_request_created( + conditional_access_responses = await evaluate_conditional_access( + ConditionalAccessHook.GROUP_REQUEST_CREATED, group_request=group_request, requester=requester, ) @@ -206,7 +204,8 @@ async def execute(self) -> Optional[GroupRequest]: return group_request # Send notification to approvers - self.notification_hook.access_group_request_created( + await send_notification( + NotificationHook.ACCESS_GROUP_REQUEST_CREATED, group_request=group_request, requester=requester, approvers=approvers, diff --git a/api/operations/create_role_request.py b/api/operations/create_role_request.py index bbf3ce51..58a91223 100644 --- a/api/operations/create_role_request.py +++ b/api/operations/create_role_request.py @@ -26,7 +26,7 @@ from api.models.tag import coalesce_constraints from api.operations.approve_role_request import ApproveRoleRequest from api.operations.reject_role_request import RejectRoleRequest -from api.plugins import get_conditional_access_hook, get_notification_hook +from api.plugins import ConditionalAccessHook, NotificationHook, evaluate_conditional_access, send_notification from api.schemas import AuditLogSchema, EventType @@ -51,9 +51,6 @@ def __init__( self.request_reason = request_reason self.request_ending_at = request_ending_at - self.conditional_access_hook = get_conditional_access_hook() - self.notification_hook = get_notification_hook() - async def execute(self) -> Optional[RoleRequest]: requester = await db.session.get(OktaUser, self.requester_user_id) @@ -184,7 +181,8 @@ async def execute(self) -> Optional[RoleRequest]: ) ) - conditional_access_responses = self.conditional_access_hook.role_request_created( + conditional_access_responses = await evaluate_conditional_access( + ConditionalAccessHook.ROLE_REQUEST_CREATED, role_request=role_request, role=requester_role, group=requested_group, @@ -211,7 +209,8 @@ async def execute(self) -> Optional[RoleRequest]: return role_request - self.notification_hook.access_role_request_created( + await send_notification( + NotificationHook.ACCESS_ROLE_REQUEST_CREATED, role_request=role_request, role=requester_role, group=requested_group, diff --git a/api/operations/delete_group.py b/api/operations/delete_group.py index a9cf94a3..5c0abc45 100644 --- a/api/operations/delete_group.py +++ b/api/operations/delete_group.py @@ -24,7 +24,7 @@ ) from api.operations.reject_access_request import RejectAccessRequest from api.operations.reject_role_request import RejectRoleRequest -from api.plugins.app_group_lifecycle import invoke_app_group_lifecycle_hook +from api.plugins.app_group_lifecycle import AppGroupLifecycleHook, invoke_app_group_lifecycle_hook from api.services import okta from api.schemas import AuditLogSchema, EventType @@ -313,6 +313,6 @@ async def execute(self) -> None: await db.session.commit() # Invoke app group lifecycle plugin hook, if configured - await invoke_app_group_lifecycle_hook("group_deleted", group=group) + await invoke_app_group_lifecycle_hook(AppGroupLifecycleHook.GROUP_DELETED, group=group) await drain_fan_out_tasks(okta_tasks, f"DeleteGroup for group {self.group_id}") diff --git a/api/operations/modify_group_details.py b/api/operations/modify_group_details.py index 95cba897..91e6044f 100644 --- a/api/operations/modify_group_details.py +++ b/api/operations/modify_group_details.py @@ -6,7 +6,7 @@ from api.extensions import db from api.models import App, AppGroup, OktaGroup, OktaUser, RoleGroup -from api.plugins.app_group_lifecycle import invoke_app_group_lifecycle_hook +from api.plugins.app_group_lifecycle import AppGroupLifecycleHook, invoke_app_group_lifecycle_hook from api.services import okta from api.schemas import AuditLogSchema, EventType @@ -85,7 +85,10 @@ async def execute(self) -> OktaGroup: # Fire group_updated hook if name or description changed if old_name != self.group.name or old_description != self.group.description: await invoke_app_group_lifecycle_hook( - "group_updated", group=self.group, old_name=old_name, old_description=old_description + AppGroupLifecycleHook.GROUP_UPDATED, + group=self.group, + old_name=old_name, + old_description=old_description, ) # Audit logging, only if group name changed diff --git a/api/operations/modify_group_type.py b/api/operations/modify_group_type.py index 6a16d439..4e7dad0b 100644 --- a/api/operations/modify_group_type.py +++ b/api/operations/modify_group_type.py @@ -20,7 +20,7 @@ from api.operations.modify_group_users import ModifyGroupUsers from api.operations.modify_groups_time_limit import ModifyGroupsTimeLimit from api.operations.modify_role_groups import ModifyRoleGroups -from api.plugins.app_group_lifecycle import invoke_app_group_lifecycle_hook +from api.plugins.app_group_lifecycle import AppGroupLifecycleHook, invoke_app_group_lifecycle_hook from api.schemas import AuditLogSchema, EventType @@ -105,7 +105,7 @@ async def execute(self) -> OktaGroup: # Invoke group_deleted hook before the AppGroup row is removed so the # plugin can still access group.app and status values (e.g. to delete # the linked GitHub team). - await invoke_app_group_lifecycle_hook("group_deleted", group=group) + await invoke_app_group_lifecycle_hook(AppGroupLifecycleHook.GROUP_DELETED, group=group) # Remove app tag map for this group that is no longer attached to an app await db.session.execute( @@ -261,7 +261,7 @@ async def execute(self) -> OktaGroup: # Invoke group_created hook after converting to an AppGroup (symmetric # with group_deleted which fires when converting away from AppGroup). if type(self.group_changes) is AppGroup: - await invoke_app_group_lifecycle_hook("group_created", group=group) + await invoke_app_group_lifecycle_hook(AppGroupLifecycleHook.GROUP_CREATED, group=group) # Audit logging if type changed if group.type != old_group_type: diff --git a/api/operations/modify_group_users.py b/api/operations/modify_group_users.py index a1d2c05d..05aeeb33 100644 --- a/api/operations/modify_group_users.py +++ b/api/operations/modify_group_users.py @@ -25,8 +25,8 @@ from api.models.access_request import get_all_possible_request_approvers from api.models.tag import coalesce_ended_at from api.operations.constraints import CheckForReason, CheckForSelfAdd -from api.plugins import get_notification_hook -from api.plugins.app_group_lifecycle import invoke_app_group_lifecycle_hook +from api.plugins import NotificationHook, send_notification +from api.plugins.app_group_lifecycle import AppGroupLifecycleHook, invoke_app_group_lifecycle_hook from api.services import okta from api.schemas import AuditLogSchema, EventType @@ -64,8 +64,6 @@ def __init__( self.created_reason = created_reason self.notify = notify - self.notification_hook = get_notification_hook() - async def execute(self) -> OktaGroup: group = ( await db.session.scalars( @@ -458,7 +456,9 @@ async def execute(self) -> OktaGroup: # Invoke app group lifecycle plugin hooks for removed members for affected_group, members in members_lost_by_group.items(): - await invoke_app_group_lifecycle_hook("group_members_removed", group=affected_group, members=members) + await invoke_app_group_lifecycle_hook( + AppGroupLifecycleHook.GROUP_MEMBERS_REMOVED, group=affected_group, members=members + ) # Commit all changes so far await db.session.commit() @@ -671,7 +671,9 @@ async def execute(self) -> OktaGroup: # Invoke app group lifecycle plugin hooks for added members for affected_group, members in members_gained_by_group.items(): - await invoke_app_group_lifecycle_hook("group_members_added", group=affected_group, members=members) + await invoke_app_group_lifecycle_hook( + AppGroupLifecycleHook.GROUP_MEMBERS_ADDED, group=affected_group, members=members + ) # Approve any pending access requests for access granted by this operation pending_requests_query = ( @@ -773,7 +775,8 @@ async def _notify_access_request( requester: Optional[OktaUser], approvers: Set[OktaUser], ) -> None: - self.notification_hook.access_request_completed( + await send_notification( + NotificationHook.ACCESS_REQUEST_COMPLETED, access_request=access_request, group=group, requester=requester, diff --git a/api/operations/modify_role_groups.py b/api/operations/modify_role_groups.py index 1864fd5d..e5d7967a 100644 --- a/api/operations/modify_role_groups.py +++ b/api/operations/modify_role_groups.py @@ -26,8 +26,12 @@ from api.models.access_request import get_all_possible_request_approvers from api.models.tag import coalesce_ended_at from api.operations.constraints import CheckForReason, CheckForSelfAdd -from api.plugins import get_notification_hook -from api.plugins.app_group_lifecycle import get_app_group_lifecycle_plugin_to_invoke, invoke_app_group_lifecycle_hook +from api.plugins import NotificationHook, send_notification +from api.plugins.app_group_lifecycle import ( + AppGroupLifecycleHook, + get_app_group_lifecycle_plugin_to_invoke, + invoke_app_group_lifecycle_hook, +) from api.services import okta from api.schemas import AuditLogSchema, EventType @@ -67,8 +71,6 @@ def __init__( self.notify = notify - self.notification_hook = get_notification_hook() - async def execute(self) -> RoleGroup: self.role = ( await db.session.scalars( @@ -310,7 +312,7 @@ async def execute(self) -> RoleGroup: ) ).all() await invoke_app_group_lifecycle_hook( - "group_members_removed", group=group, members=members_losing_access + AppGroupLifecycleHook.GROUP_MEMBERS_REMOVED, group=group, members=members_losing_access ) if self.sync_to_okta: @@ -451,7 +453,7 @@ async def execute(self) -> RoleGroup: ) ).all() await invoke_app_group_lifecycle_hook( - "group_members_added", group=group, members=members_gaining_access + AppGroupLifecycleHook.GROUP_MEMBERS_ADDED, group=group, members=members_gaining_access ) for member in active_role_memberships: @@ -705,7 +707,8 @@ async def _notify_access_request( requester: Optional[OktaUser], approvers: Set[OktaUser], ) -> None: - self.notification_hook.access_request_completed( + await send_notification( + NotificationHook.ACCESS_REQUEST_COMPLETED, access_request=access_request, group=group, requester=requester, @@ -731,7 +734,8 @@ async def _notify_role_request( requester: Optional[OktaUser], approvers: Set[OktaUser], ) -> None: - self.notification_hook.access_role_request_completed( + await send_notification( + NotificationHook.ACCESS_ROLE_REQUEST_COMPLETED, role_request=role_request, role=role, group=group, diff --git a/api/operations/reject_access_request.py b/api/operations/reject_access_request.py index b30bc2c5..11ba0eba 100644 --- a/api/operations/reject_access_request.py +++ b/api/operations/reject_access_request.py @@ -10,7 +10,7 @@ from api.extensions import db from api.models import AccessRequest, AccessRequestStatus, AppGroup, OktaGroup, OktaUser, RoleGroup from api.models.access_request import get_all_possible_request_approvers -from api.plugins import get_notification_hook +from api.plugins import NotificationHook, send_notification from api.schemas import AuditLogSchema, EventType @@ -35,8 +35,6 @@ def __init__( self.notify = notify self.notify_requester = notify_requester - self.notification_hook = get_notification_hook() - async def execute(self) -> AccessRequest: # Lock the request row so a reject can't race a concurrent approve/ # reject; both serialize on this row and the loser hits the resolved @@ -109,7 +107,8 @@ async def execute(self) -> AccessRequest: approvers = await get_all_possible_request_approvers(access_request) - self.notification_hook.access_request_completed( + await send_notification( + NotificationHook.ACCESS_REQUEST_COMPLETED, access_request=access_request, group=group, requester=requester, diff --git a/api/operations/reject_group_request.py b/api/operations/reject_group_request.py index 52169c2e..78058866 100644 --- a/api/operations/reject_group_request.py +++ b/api/operations/reject_group_request.py @@ -10,7 +10,7 @@ from api.models import AccessRequestStatus, AppGroup, GroupRequest, OktaUser, OktaUserGroupMember from api.models.access_request import get_all_possible_request_approvers from api.models.app_group import get_access_owners -from api.plugins import get_notification_hook +from api.plugins import NotificationHook, send_notification from api.schemas import AuditLogSchema, EventType @@ -35,8 +35,6 @@ def __init__( self.notify = notify self.notify_requester = notify_requester - self.notification_hook = get_notification_hook() - async def execute(self) -> GroupRequest: # Lock the request row so a reject can't race a concurrent approve/ # reject; both serialize on this row and the loser hits the resolved @@ -126,7 +124,8 @@ async def execute(self) -> GroupRequest: approvers = await get_all_possible_request_approvers(group_request) - self.notification_hook.access_group_request_completed( + await send_notification( + NotificationHook.ACCESS_GROUP_REQUEST_COMPLETED, group_request=group_request, group=None, requester=requester, diff --git a/api/operations/reject_role_request.py b/api/operations/reject_role_request.py index f011374f..5315d37b 100644 --- a/api/operations/reject_role_request.py +++ b/api/operations/reject_role_request.py @@ -10,7 +10,7 @@ from api.extensions import db from api.models import AccessRequestStatus, AppGroup, OktaGroup, OktaUser, RoleRequest from api.models.access_request import get_all_possible_request_approvers -from api.plugins import get_notification_hook +from api.plugins import NotificationHook, send_notification from api.schemas import AuditLogSchema, EventType @@ -35,8 +35,6 @@ def __init__( self.notify = notify self.notify_requester = notify_requester - self.notification_hook = get_notification_hook() - async def execute(self) -> RoleRequest: # Lock the request row so a reject can't race a concurrent approve/ # reject; both serialize on this row and the loser hits the resolved @@ -119,7 +117,8 @@ async def execute(self) -> RoleRequest: approvers = await get_all_possible_request_approvers(role_request) - self.notification_hook.access_role_request_completed( + await send_notification( + NotificationHook.ACCESS_ROLE_REQUEST_COMPLETED, role_request=role_request, role=requester_role, group=group, diff --git a/api/plugins/__init__.py b/api/plugins/__init__.py index 8d34227e..25e688af 100644 --- a/api/plugins/__init__.py +++ b/api/plugins/__init__.py @@ -1,6 +1,7 @@ import pluggy from api.plugins.app_group_lifecycle import ( + AppGroupLifecycleHook, AppGroupLifecyclePluginConfigProperty, AppGroupLifecyclePluginFilteringError, AppGroupLifecyclePluginMetadata, @@ -21,9 +22,14 @@ validate_app_group_lifecycle_plugin_app_config, validate_app_group_lifecycle_plugin_group_config, ) -from api.plugins.conditional_access import ConditionalAccessResponse, get_conditional_access_hook +from api.plugins.conditional_access import ( + ConditionalAccessHook, + ConditionalAccessResponse, + evaluate_conditional_access, + get_conditional_access_hook, +) from api.plugins.metrics_reporter import get_metrics_reporter_hook -from api.plugins.notifications import get_notification_hook +from api.plugins.notifications import NotificationHook, get_notification_hook, send_notification app_group_lifecycle_hook_impl = pluggy.HookimplMarker("access_app_group_lifecycle") conditional_access_hook_impl = pluggy.HookimplMarker("access_conditional_access") @@ -47,6 +53,7 @@ def load_plugins() -> None: __all__ = [ # App Group Lifecycle Plugin "app_group_lifecycle_plugin_name", + "AppGroupLifecycleHook", "AppGroupLifecyclePluginConfigProperty", "AppGroupLifecyclePluginFilteringError", "AppGroupLifecyclePluginMetadata", @@ -67,11 +74,15 @@ def load_plugins() -> None: "validate_app_group_lifecycle_plugin_group_config", "app_group_lifecycle_hook_impl", # Conditional Access Plugin + "ConditionalAccessHook", "ConditionalAccessResponse", "get_conditional_access_hook", + "evaluate_conditional_access", "conditional_access_hook_impl", # Notifications Plugin + "NotificationHook", "get_notification_hook", + "send_notification", "notification_hook_impl", # Metrics Reporter Plugin "get_metrics_reporter_hook", diff --git a/api/plugins/_async_dispatch.py b/api/plugins/_async_dispatch.py new file mode 100644 index 00000000..eb664996 --- /dev/null +++ b/api/plugins/_async_dispatch.py @@ -0,0 +1,92 @@ +"""Shared helpers for the async pluggy plugin interface. + +`pluggy` never awaits a hook: calling ``pm.hook.(**kwargs)`` just invokes +each registered implementation and collects its return value. For the Access 2.0 +async plugin interface the implementations are ``async def``, so the call returns +a list of coroutines that the application runs to completion via +``run_hooks_to_completion``. + +A stale *synchronous* implementation of an async hook would return a plain value +instead of a coroutine, which then fails when scheduled as a task. +``verify_async_impls`` turns that into a clear, load-time error naming the +offending plugin — mirroring the fail-fast spirit of +``metrics_reporter._verify_tag_forwarding``. +""" + +import asyncio +import inspect +import logging +from typing import Any + +import pluggy + +logger = logging.getLogger("api") + + +def verify_async_impls(pm: pluggy.PluginManager, hook_names: tuple[str, ...]) -> None: + """Raise if any registered impl for an async hook is not a coroutine function. + + Args: + pm: The plugin manager whose registered implementations to inspect. + hook_names: The names of the hooks that the application awaits. Hooks that + are intentionally synchronous (metadata/config/status schemas) or that + return an async context manager (``batch_metrics``) must be excluded by + the caller. + """ + offenders: list[str] = [] + for hook_name in hook_names: + caller = getattr(pm.hook, hook_name, None) + if caller is None: + continue + for impl in caller.get_hookimpls(): + if not inspect.iscoroutinefunction(impl.function): + offenders.append(f"{pm.get_name(impl.plugin)}.{hook_name}") + if offenders: + raise RuntimeError( + "The Access 2.0 plugin interface requires these hook implementations to be " + "declared with 'async def', but they are synchronous: " + "; ".join(sorted(offenders)) + ) + + +async def run_hooks_to_completion(coros: list[Any], *, context: str) -> tuple[list[Any], list[BaseException]]: + """Run the coroutines a pluggy hook call returned to completion. + + Returns ``(results, exceptions)``: the return values of the implementations + that succeeded (in input order) and the exceptions raised by those that + failed. Every failure is also logged at ERROR with its traceback — the app + does not swallow plugin errors, it surfaces them; a plugin that expects noisy + failures (e.g. connection timeouts) should catch those itself. The call still + never raises on a plugin's behalf, so one plugin can't break the operation. + + Uses ``asyncio.wait`` rather than ``asyncio.gather`` deliberately, mirroring + the fan-out drain in ``api/operations/_fan_out.py`` (discord/access#481): + + - If the awaiting caller is cancelled (e.g. the client disconnected), + ``gather`` propagates the cancellation to its children and tears down the + in-flight hook coroutines; ``wait`` leaves them running to completion. + These hooks fire after the authoritative DB change has committed and only + do network I/O (notifications, metrics), so an in-flight send must be + allowed to finish. + - One implementation failing must never cancel or abandon its siblings — + ``wait`` runs them all and lets us collect every result and every error. + + Keep this ``wait``; do not "simplify" it back to ``gather``. + """ + tasks = [asyncio.ensure_future(coro) for coro in coros] + if not tasks: + # No implementations registered (or a test double yielded nothing). + # asyncio.wait() rejects an empty set, so short-circuit. + return [], [] + + await asyncio.wait(tasks) + + results: list[Any] = [] + exceptions: list[BaseException] = [] + for task in tasks: + exc = task.exception() + if exc is None: + results.append(task.result()) + else: + exceptions.append(exc) + logger.error("%s: plugin hook raised %r", context, exc, exc_info=exc) + return results, exceptions diff --git a/api/plugins/app_group_lifecycle.py b/api/plugins/app_group_lifecycle.py index d7d11ef5..5c0b8fcd 100644 --- a/api/plugins/app_group_lifecycle.py +++ b/api/plugins/app_group_lifecycle.py @@ -1,13 +1,14 @@ import logging -import sys from dataclasses import asdict, dataclass +from enum import StrEnum from typing import Any, Literal import pluggy -from sqlalchemy.orm import Session +from sqlalchemy.ext.asyncio import AsyncSession from api.extensions import db from api.models import App, AppGroup, OktaUser +from api.plugins._async_dispatch import run_hooks_to_completion, verify_async_impls app_group_lifecycle_plugin_name = "access_app_group_lifecycle" hookspec = pluggy.HookspecMarker(app_group_lifecycle_plugin_name) @@ -16,6 +17,20 @@ _cached_app_group_lifecycle_hook: pluggy.HookRelay | None = None +class AppGroupLifecycleHook(StrEnum): + """The lifecycle hooks that receive an AsyncSession and are awaited by the + application (StrEnum value == the pluggy hook name). The + metadata/config/status/validation hooks are pure schema accessors and remain + synchronous, so they are not members here (and are excluded from the async check).""" + + GROUP_CREATED = "group_created" + GROUP_UPDATED = "group_updated" + GROUP_DELETED = "group_deleted" + GROUP_MEMBERS_ADDED = "group_members_added" + GROUP_MEMBERS_REMOVED = "group_members_removed" + SYNC_ALL_GROUP_MEMBERSHIP = "sync_all_group_membership" + + class PluginNotFoundError(Exception): """Raised by plugin endpoints when the requested plugin id is not registered. The exception handler in `api/exception_handlers.py` @@ -161,26 +176,26 @@ def get_plugin_group_status_properties( # Group lifecycle hooks @hookspec - def group_created(self, session: Session, group: AppGroup, plugin_id: str | None) -> None: + async def group_created(self, session: AsyncSession, group: AppGroup, plugin_id: str | None) -> None: """ Handle group creation. Args: - session: The Access database session. + session: The Access database AsyncSession. Await ORM calls on it. group: The app group that was created. plugin_id: If provided, only the plugin matching this ID should respond. If None, all plugins may respond. """ @hookspec - def group_updated( - self, session: Session, group: AppGroup, old_name: str, old_description: str, plugin_id: str | None + async def group_updated( + self, session: AsyncSession, group: AppGroup, old_name: str, old_description: str, plugin_id: str | None ) -> None: """ Handle group update (name or description change). Args: - session: The Access database session. + session: The Access database AsyncSession. Await ORM calls on it. group: The app group after the update. old_name: The group's name before the update. old_description: The group's description before the update. @@ -189,12 +204,12 @@ def group_updated( """ @hookspec - def group_deleted(self, session: Session, group: AppGroup, plugin_id: str | None) -> None: + async def group_deleted(self, session: AsyncSession, group: AppGroup, plugin_id: str | None) -> None: """ Handle group deletion. Args: - session: The Access database session. + session: The Access database AsyncSession. Await ORM calls on it. group: The app group that was deleted. plugin_id: If provided, only the plugin matching this ID should respond. If None, all plugins may respond. @@ -203,14 +218,14 @@ def group_deleted(self, session: Session, group: AppGroup, plugin_id: str | None # Membership hooks @hookspec - def group_members_added( - self, session: Session, group: AppGroup, members: list[OktaUser], plugin_id: str | None + async def group_members_added( + self, session: AsyncSession, group: AppGroup, members: list[OktaUser], plugin_id: str | None ) -> None: """ Handle member addition. Args: - session: The Access database session. + session: The Access database AsyncSession. Await ORM calls on it. group: The app group to which members were added. members: The list of users that were added to the group. plugin_id: If provided, only the plugin matching this ID should respond. @@ -218,14 +233,14 @@ def group_members_added( """ @hookspec - def group_members_removed( - self, session: Session, group: AppGroup, members: list[OktaUser], plugin_id: str | None + async def group_members_removed( + self, session: AsyncSession, group: AppGroup, members: list[OktaUser], plugin_id: str | None ) -> None: """ Handle member removal. Args: - session: The Access database session. + session: The Access database AsyncSession. Await ORM calls on it. group: The app group from which members were removed. members: The list of users that were removed from the group. plugin_id: If provided, only the plugin matching this ID should respond. @@ -233,12 +248,12 @@ def group_members_removed( """ @hookspec - def sync_all_group_membership(self, session: Session, app: App, plugin_id: str | None) -> None: + async def sync_all_group_membership(self, session: AsyncSession, app: App, plugin_id: str | None) -> None: """ - Bulk sync all group memberships for an app. This is invoked periodically by a CLI command `flask sync-app-group-memberships`. + Bulk sync all group memberships for an app. This is invoked periodically by the CLI command `access sync-app-group-memberships`. Args: - session: The Access database session. + session: The Access database AsyncSession. Await ORM calls on it. app: The app for which to sync all group membership. plugin_id: If provided, only the plugin matching this ID should respond. If None, all plugins may respond. @@ -255,11 +270,9 @@ def get_app_group_lifecycle_hook() -> pluggy.HookRelay: pm = pluggy.PluginManager(app_group_lifecycle_plugin_name) pm.add_hookspecs(AppGroupLifecyclePluginSpec) - # Register the hook wrappers - pm.register(sys.modules[__name__]) - count = pm.load_setuptools_entrypoints(app_group_lifecycle_plugin_name) logger.info(f"Loaded {count} app group lifecycle plugin(s)") + verify_async_impls(pm, tuple(AppGroupLifecycleHook)) _cached_app_group_lifecycle_hook = pm.hook @@ -331,14 +344,14 @@ def get_app_group_lifecycle_plugin_to_invoke(group: Any) -> str | None: return group.app.app_group_lifecycle_plugin -async def invoke_app_group_lifecycle_hook(hook_method: str, *, group: Any, **kwargs: Any) -> None: +async def invoke_app_group_lifecycle_hook(hook_method: AppGroupLifecycleHook, *, group: Any, **kwargs: Any) -> None: """Invoke an app-group lifecycle hook for ``group``, if a plugin is configured. - No-op when no lifecycle plugin applies to ``group``. The hookspecs take a sync - ``Session``, so this bridges the request's ``AsyncSession`` through ``run_sync`` - and hands the plugin a working sync session (TODO 18). Commits on success; on any - hook error it logs and rolls back so a misbehaving plugin can't abort the - surrounding operation. + No-op when no lifecycle plugin applies to ``group``. The lifecycle hooks are + native async: they receive the request's ``AsyncSession`` directly + and run on the event loop, so no ``run_sync`` bridge is needed. Commits on + success; on any hook error it logs and rolls back so a misbehaving plugin + can't abort the surrounding operation. ``kwargs`` are forwarded to the hook alongside ``session`` and ``group`` — e.g. ``members=`` for the membership hooks, ``old_name=``/``old_description=`` for @@ -347,15 +360,24 @@ async def invoke_app_group_lifecycle_hook(hook_method: str, *, group: Any, **kwa plugin_id = get_app_group_lifecycle_plugin_to_invoke(group) if plugin_id is None: return + hook = get_app_group_lifecycle_hook() + # run_hooks_to_completion uses asyncio.wait (not gather): a cancelled request + # won't tear down an in-flight hook, and one plugin failing won't cancel the + # others. Failures are logged there; we roll back rather than commit partial + # writes, but never propagate, so a misbehaving plugin can't abort the + # surrounding operation. + _, exceptions = await run_hooks_to_completion( + getattr(hook, hook_method)(session=db.session, group=group, plugin_id=plugin_id, **kwargs), + context=f"{hook_method} hook for group {getattr(group, 'id', None)} (plugin '{plugin_id}')", + ) + if exceptions: + await db.session.rollback() + return try: - hook = get_app_group_lifecycle_hook() - await db.session.run_sync( - lambda s: getattr(hook, hook_method)(session=s, group=group, plugin_id=plugin_id, **kwargs) - ) await db.session.commit() except Exception: logging.getLogger("api").exception( - f"Failed to invoke {hook_method} hook for group {getattr(group, 'id', None)} with plugin '{plugin_id}'" + f"Failed to commit after {hook_method} hook for group {getattr(group, 'id', None)} with plugin '{plugin_id}'" ) await db.session.rollback() diff --git a/api/plugins/conditional_access.py b/api/plugins/conditional_access.py index da77a413..4c048ac0 100644 --- a/api/plugins/conditional_access.py +++ b/api/plugins/conditional_access.py @@ -1,12 +1,13 @@ import logging -import sys from dataclasses import dataclass from datetime import datetime -from typing import Any, Generator, List, Optional +from enum import StrEnum +from typing import Any, List, Optional import pluggy from api.models import AccessRequest, App, GroupRequest, OktaGroup, OktaUser, RoleGroup, RoleRequest, Tag +from api.plugins._async_dispatch import run_hooks_to_completion, verify_async_impls conditional_access_plugin_name = "access_conditional_access" hookspec = pluggy.HookspecMarker(conditional_access_plugin_name) @@ -17,6 +18,14 @@ logger = logging.getLogger(__name__) +class ConditionalAccessHook(StrEnum): + """Conditional-access hook names (StrEnum value == the pluggy hook name).""" + + ACCESS_REQUEST_CREATED = "access_request_created" + ROLE_REQUEST_CREATED = "role_request_created" + GROUP_REQUEST_CREATED = "group_request_created" + + @dataclass class ConditionalAccessResponse: approved: bool @@ -26,19 +35,19 @@ class ConditionalAccessResponse: class ConditionalAccessPluginSpec: @hookspec - def access_request_created( + async def access_request_created( self, access_request: AccessRequest, group: OktaGroup, group_tags: List[Tag], requester: OktaUser ) -> Optional[ConditionalAccessResponse]: """Automatically approve, deny, or continue the access request.""" @hookspec - def role_request_created( + async def role_request_created( self, role_request: RoleRequest, role: RoleGroup, group: OktaGroup, group_tags: List[Tag], requester: OktaUser ) -> Optional[ConditionalAccessResponse]: """Automatically approve, deny, or continue the role request.""" @hookspec - def group_request_created( + async def group_request_created( self, group_request: GroupRequest, requester: OktaUser, @@ -47,52 +56,27 @@ def group_request_created( """Automatically approve, deny, or continue the group request.""" -@hookimpl(wrapper=True) -def access_request_created( - access_request: AccessRequest, group: OktaGroup, group_tags: List[Tag], requester: OktaUser -) -> Generator[Any, None, Optional[ConditionalAccessResponse]] | List[Optional[ConditionalAccessResponse]]: - try: - # Trigger exception if it exists - return (yield) - except Exception: - # Log and do not raise since request failures should not - # break the flow. The access request can still be manually - # approved or denied - logger.exception("Failed to execute access request created callback") - - return [] - - -@hookimpl(wrapper=True) -def role_request_created( - role_request: RoleRequest, role: RoleGroup, group: OktaGroup, group_tags: List[Tag], requester: OktaUser -) -> Generator[Any, None, Optional[ConditionalAccessResponse]] | List[Optional[ConditionalAccessResponse]]: - try: - # Trigger exception if it exists - return (yield) - except Exception: - # Log and do not raise since request failures should not - # break the flow. The access request can still be manually - # approved or denied - logger.exception("Failed to execute role request created callback") - - return [] - - -@hookimpl(wrapper=True) -def group_request_created( - group_request: GroupRequest, requester: OktaUser, app: Optional[App] = None -) -> Generator[Any, None, Optional[ConditionalAccessResponse]] | List[Optional[ConditionalAccessResponse]]: - try: - # Trigger exception if it exists - return (yield) - except Exception: - # Log and do not raise since request failures should not - # break the flow. The access request can still be manually - # approved or denied - logger.exception("Failed to execute group request created callback") - - return [] +async def evaluate_conditional_access( + hook: ConditionalAccessHook, /, **kwargs: Any +) -> list[Optional[ConditionalAccessResponse]]: + """Run an async conditional-access hook and return each plugin's response. + + This is the async replacement for the old ``@hookimpl(wrapper=True)`` + wrappers: pluggy cannot wrap coroutines, so the run + error handling lives + here. Implementations run via ``run_hooks_to_completion`` (``asyncio.wait``, + not ``gather``) so one plugin's failure doesn't cancel the others; a failing + plugin is logged and simply omitted from the responses, leaving that decision + to manual approval/denial rather than breaking the request. + + ``kwargs`` are forwarded verbatim; pluggy passes each implementation only the + parameters it declares, so extra kwargs (e.g. ``requester_role``) are ignored + safely. + """ + relay = get_conditional_access_hook() + results, _ = await run_hooks_to_completion( + getattr(relay, hook)(**kwargs), context=f"{hook} conditional access callback" + ) + return list(results) def get_conditional_access_hook() -> pluggy.HookRelay: @@ -104,11 +88,9 @@ def get_conditional_access_hook() -> pluggy.HookRelay: pm = pluggy.PluginManager(conditional_access_plugin_name) pm.add_hookspecs(ConditionalAccessPluginSpec) - # Register the hook wrappers - pm.register(sys.modules[__name__]) - count = pm.load_setuptools_entrypoints(conditional_access_plugin_name) print(f"Count of loaded conditional access plugins: {count}") + verify_async_impls(pm, tuple(ConditionalAccessHook)) _cached_conditional_access_hook = pm.hook return _cached_conditional_access_hook diff --git a/api/plugins/metrics_reporter.py b/api/plugins/metrics_reporter.py index 1b227ace..69f7e2e8 100644 --- a/api/plugins/metrics_reporter.py +++ b/api/plugins/metrics_reporter.py @@ -1,10 +1,11 @@ import inspect import logging -import sys -from typing import ContextManager, Dict, List, Optional +from typing import AsyncContextManager, Dict, List, Optional import pluggy +from api.plugins._async_dispatch import verify_async_impls + metrics_reporter_plugin_name = "access_metrics_reporter" hookspec = pluggy.HookspecMarker(metrics_reporter_plugin_name) hookimpl = pluggy.HookimplMarker(metrics_reporter_plugin_name) @@ -17,12 +18,17 @@ _TAG_FORWARDING_HOOKS = ("record_counter", "record_gauge", "record_histogram", "record_summary") _MUST_NOT_DEFAULT = ("value", "tags") +# Awaited metric hooks — must be `async def`. `batch_metrics` is +# excluded: it returns an async context manager (used via `async with`), not a +# coroutine, so it would fail the coroutine-function check. +_ASYNC_HOOKS = ("record_counter", "record_gauge", "record_histogram", "record_summary", "set_global_tags", "flush") + logger = logging.getLogger(__name__) class MetricsReporterPluginSpec: @hookspec - def record_counter( + async def record_counter( self, metric_name: str, value: float, @@ -43,7 +49,7 @@ def record_counter( """ @hookspec - def record_gauge( + async def record_gauge( self, metric_name: str, value: float, @@ -52,7 +58,7 @@ def record_gauge( """Record a gauge metric value (snapshot/current value).""" @hookspec - def record_histogram( + async def record_histogram( self, metric_name: str, value: float, @@ -72,7 +78,7 @@ def record_histogram( """ @hookspec - def record_summary( + async def record_summary( self, metric_name: str, value: float, @@ -86,31 +92,32 @@ def record_summary( """ @hookspec - def batch_metrics(self) -> ContextManager[None]: + def batch_metrics(self) -> AsyncContextManager[None]: """ - Context manager for batching multiple metric operations. + Async context manager for batching multiple metric operations. - Returns a context manager that batches metric operations for efficiency. - Particularly useful for HTTP-based backends to reduce network calls. + Returns an async context manager that batches metric operations for + efficiency. Particularly useful for HTTP-based backends to reduce network + calls. Example: - with metrics.batch_metrics(): - metrics.record_counter("requests", 1, tags={"method": "GET"}) - metrics.record_gauge("queue_size", 42, tags=None) - metrics.record_histogram("response_time", 0.123, tags={"route": "/api"}) + async with metrics.batch_metrics(): + await metrics.record_counter("requests", 1, tags={"method": "GET"}) + await metrics.record_gauge("queue_size", 42, tags=None) + await metrics.record_histogram("response_time", 0.123, tags={"route": "/api"}) # All metrics sent in one batch here """ return NotImplemented @hookspec - def set_global_tags( + async def set_global_tags( self, tags: Dict[str, str], ) -> None: """Set global tags to be included with all metrics.""" @hookspec - def flush(self) -> None: + async def flush(self) -> None: """Force flush any buffered metrics to the backend.""" @@ -152,12 +159,10 @@ def get_metrics_reporter_hook() -> pluggy.HookRelay: pm = pluggy.PluginManager(metrics_reporter_plugin_name) pm.add_hookspecs(MetricsReporterPluginSpec) - # Register the hook wrappers - pm.register(sys.modules[__name__]) - count = pm.load_setuptools_entrypoints(metrics_reporter_plugin_name) logger.debug(f"Count of loaded metrics reporter plugins: {count}") _verify_tag_forwarding(pm) + verify_async_impls(pm, _ASYNC_HOOKS) _cached_metrics_reporter_hook = pm.hook return _cached_metrics_reporter_hook diff --git a/api/plugins/notifications.py b/api/plugins/notifications.py index 7f5d9eb2..3b966788 100644 --- a/api/plugins/notifications.py +++ b/api/plugins/notifications.py @@ -1,7 +1,7 @@ import datetime import logging -import sys -from typing import Dict, Generator, Optional +from enum import StrEnum +from typing import Any, Dict, Optional import pluggy @@ -15,6 +15,7 @@ RoleGroupMap, RoleRequest, ) +from api.plugins._async_dispatch import run_hooks_to_completion, verify_async_impls from api.plugins.metrics_reporter import get_metrics_reporter_hook notification_plugin_name = "access_notifications" @@ -26,22 +27,55 @@ logger = logging.getLogger(__name__) -def _record_sent(metric_name: str, tags: Optional[Dict[str, str]] = None) -> None: +class NotificationHook(StrEnum): + """Notification hook names. Each value is the pluggy hook attribute name, so a + member can be passed straight to ``getattr(hook_relay, member)``.""" + + ACCESS_REQUEST_CREATED = "access_request_created" + ACCESS_REQUEST_COMPLETED = "access_request_completed" + ACCESS_EXPIRING_USER = "access_expiring_user" + ACCESS_EXPIRING_OWNER = "access_expiring_owner" + ACCESS_EXPIRING_ROLE_OWNER = "access_expiring_role_owner" + ACCESS_ROLE_REQUEST_CREATED = "access_role_request_created" + ACCESS_ROLE_REQUEST_COMPLETED = "access_role_request_completed" + ACCESS_GROUP_REQUEST_CREATED = "access_group_request_created" + ACCESS_GROUP_REQUEST_COMPLETED = "access_group_request_completed" + + +# hook -> (metric name, static tags) recorded once the hook fans out +# successfully. Kept here (not at call sites) so the "sent" accounting lives +# next to the spec it measures. +_SENT_METRICS: dict[NotificationHook, tuple[str, Optional[Dict[str, str]]]] = { + NotificationHook.ACCESS_REQUEST_CREATED: ("notifications.access_request_created.sent", None), + NotificationHook.ACCESS_REQUEST_COMPLETED: ("notifications.access_request_completed.sent", None), + NotificationHook.ACCESS_EXPIRING_USER: ("notifications.expiring_access.sent", {"kind": "user"}), + NotificationHook.ACCESS_EXPIRING_OWNER: ("notifications.expiring_access.sent", {"kind": "owner"}), + NotificationHook.ACCESS_EXPIRING_ROLE_OWNER: ("notifications.expiring_access.sent", {"kind": "role_owner"}), + NotificationHook.ACCESS_ROLE_REQUEST_CREATED: ("notifications.role_request_created.sent", None), + NotificationHook.ACCESS_ROLE_REQUEST_COMPLETED: ("notifications.role_request_completed.sent", None), + NotificationHook.ACCESS_GROUP_REQUEST_CREATED: ("notifications.group_request_created.sent", None), + NotificationHook.ACCESS_GROUP_REQUEST_COMPLETED: ("notifications.group_request_completed.sent", None), +} + + +async def _record_sent(metric_name: str, tags: Optional[Dict[str, str]] = None) -> None: try: - get_metrics_reporter_hook().record_counter(metric_name=metric_name, value=1, tags=tags) + coros = get_metrics_reporter_hook().record_counter(metric_name=metric_name, value=1, tags=tags) except Exception: logger.exception("Failed to record %s metric", metric_name) + return + await run_hooks_to_completion(coros, context=f"metrics record_counter {metric_name}") class NotificationPluginSpec: @hookspec - def access_request_created( + async def access_request_created( self, access_request: AccessRequest, group: OktaGroup, requester: OktaUser, approvers: list[OktaUser] ) -> None: """Notify the approvers of the access request.""" @hookspec - def access_request_completed( + async def access_request_completed( self, access_request: AccessRequest, group: OktaGroup, @@ -51,45 +85,19 @@ def access_request_completed( ) -> None: """Notify the requester that their access request has been processed.""" - @hookspec( - warn_on_impl_args={ - "groups": DeprecationWarning( - "The groups parameter of access_expiring_user is deprecated and will be removed soon; " - "use okta_user_group_members instead" - ), - }, - ) - def access_expiring_user( + @hookspec + async def access_expiring_user( self, - groups: list[OktaGroup], user: OktaUser, expiration_datetime: datetime.datetime, okta_user_group_members: Optional[list[OktaUserGroupMember]], ) -> None: """Notify individuals that their access to a group is expiring soon""" - @hookspec( - warn_on_impl_args={ - "groups": DeprecationWarning( - "The groups parameter of access_expiring_owner is deprecated and will be removed soon; " - "use group_user_associations and role_group_associations instead" - ), - "roles": DeprecationWarning( - "The roles parameter of access_expiring_owner is deprecated and will be removed soon; " - "use role_group_associations instead" - ), - "users": DeprecationWarning( - "The users parameter of access_expiring_owner is deprecated and will be removed soon; " - "use group_user_associations instead" - ), - }, - ) - def access_expiring_owner( + @hookspec + async def access_expiring_owner( self, owner: OktaUser, - groups: list[OktaGroup], - roles: list[RoleGroup], - users: list[OktaUser], expiration_datetime: datetime.datetime, group_user_associations: Optional[list[OktaUserGroupMember]], role_group_associations: Optional[list[RoleGroupMap]], @@ -97,7 +105,7 @@ def access_expiring_owner( """Notify group owners that individuals or roles access to a group is expiring soon""" @hookspec - def access_expiring_role_owner( + async def access_expiring_role_owner( self, owner: OktaUser, roles: list[RoleGroupMap], @@ -106,7 +114,7 @@ def access_expiring_role_owner( """Notify role owners that roles they own will be losing access soon""" @hookspec - def access_role_request_created( + async def access_role_request_created( self, role_request: RoleRequest, role: RoleGroup, @@ -117,7 +125,7 @@ def access_role_request_created( """Notify the approvers of the role request.""" @hookspec - def access_role_request_completed( + async def access_role_request_completed( self, role_request: RoleRequest, role: RoleGroup, @@ -129,7 +137,7 @@ def access_role_request_completed( """Notify the requester that their role request has been processed.""" @hookspec - def access_group_request_created( + async def access_group_request_created( self, group_request: GroupRequest, requester: OktaUser, @@ -138,7 +146,7 @@ def access_group_request_created( """Notify the approvers of the group request.""" @hookspec - def access_group_request_completed( + async def access_group_request_completed( self, group_request: GroupRequest, group: Optional[OktaGroup], @@ -149,178 +157,33 @@ def access_group_request_completed( """Notify the requester that their group request has been processed.""" -@hookimpl(wrapper=True) -def access_request_created( - access_request: AccessRequest, group: OktaGroup, requester: OktaUser, approvers: list[OktaUser] -) -> Generator[None, None, None]: - try: - result = yield - except Exception: - # Log and do not raise since notification failures should not - # break the flow. Users can still manually ping approvers - # to process their request from the UI - logger.exception("Failed to execute access request created notification callback") - return - _record_sent("notifications.access_request_created.sent") - return result - - -@hookimpl(wrapper=True) -def access_request_completed( - access_request: AccessRequest, - group: OktaGroup, - requester: OktaUser, - approvers: list[OktaUser], - notify_requester: bool, -) -> Generator[None, None, None]: - try: - result = yield - except Exception: - # Log and do not raise since notification failures should not - # break the flow. Users can still manually ping approvers - # to process their request from the UI - logger.exception("Failed to execute access request completed notification callback") - return - _record_sent("notifications.access_request_completed.sent") - return result - - -@hookimpl(wrapper=True) -def access_expiring_user( - groups: list[OktaGroup], - user: OktaUser, - expiration_datetime: datetime.datetime, - okta_user_group_members: Optional[list[OktaUserGroupMember]], -) -> Generator[None, None, None]: - try: - result = yield - except Exception: - # Log and do not raise since notification failures should not - # break the flow. Users can still manually ping approvers - # to process their request from the UI - logger.exception("Failed to execute access expiring for user notification callback") - return - _record_sent("notifications.expiring_access.sent", tags={"kind": "user"}) - return result - - -@hookimpl(wrapper=True) -def access_expiring_owner( - owner: OktaUser, - groups: list[OktaGroup], - roles: list[RoleGroup], - users: list[OktaUser], - expiration_datetime: datetime.datetime, - group_user_associations: Optional[list[OktaUserGroupMember]], - role_group_associations: Optional[list[RoleGroupMap]], -) -> Generator[None, None, None]: - try: - result = yield - except Exception: - # Log and do not raise since notification failures should not - # break the flow. Users can still manually ping approvers - # to process their request from the UI - logger.exception("Failed to execute access expiring for owner notification callback") - return - _record_sent("notifications.expiring_access.sent", tags={"kind": "owner"}) - return result - +async def send_notification(hook: NotificationHook, /, **kwargs: Any) -> None: + """Fire an async notification hook, swallow plugin errors, and record a + "sent" counter when every implementation succeeded. -@hookimpl(wrapper=True) -def access_expiring_role_owner( - owner: OktaUser, - roles: list[RoleGroupMap], - expiration_datetime: datetime.datetime, -) -> Generator[None, None, None]: - try: - result = yield - except Exception: - # Log and do not raise since notification failures should not - # break the flow. Users can still manually ping approvers - # to process their request from the UI - logger.exception("Failed to execute access expiring for role owner notification callback") - return - _record_sent("notifications.expiring_access.sent", tags={"kind": "role_owner"}) - return result - - -@hookimpl(wrapper=True) -def access_role_request_created( - role_request: RoleRequest, - role: RoleGroup, - group: OktaGroup, - requester: OktaUser, - approvers: list[OktaUser], -) -> Generator[None, None, None]: - try: - result = yield - except Exception: - # Log and do not raise since notification failures should not - # break the flow. Users can still manually ping approvers - # to process their request from the UI - logger.exception("Failed to execute role request created notification callback") - return - _record_sent("notifications.role_request_created.sent") - return result - - -@hookimpl(wrapper=True) -def access_role_request_completed( - role_request: RoleRequest, - role: RoleGroup, - group: OktaGroup, - requester: OktaUser, - approvers: list[OktaUser], - notify_requester: bool, -) -> Generator[None, None, None]: - try: - result = yield - except Exception: - # Log and do not raise since notification failures should not - # break the flow. Users can still manually ping approvers - # to process their request from the UI - logger.exception("Failed to execute role request completed notification callback") - return - _record_sent("notifications.role_request_completed.sent") - return result + This is the async replacement for the old ``@hookimpl(wrapper=True)`` + wrappers: pluggy cannot wrap coroutines, so exception handling and the + success metric live here instead. Notification failures are logged and never + propagate — a request must still succeed even if a plugin's DM/email fails, + and approvers can be pinged manually from the UI. + The plugin coroutines are run via ``run_hooks_to_completion`` (``asyncio.wait``, + not ``gather``) so a cancelled request doesn't tear down an in-flight send and + one plugin's failure doesn't cancel the others. -@hookimpl(wrapper=True) -def access_group_request_created( - group_request: GroupRequest, - requester: OktaUser, - approvers: list[OktaUser], -) -> Generator[None, None, None]: - try: - result = yield - except Exception: - # Log and do not raise since notification failures should not - # break the flow. Users can still manually ping approvers - # to process their request from the UI - logger.exception("Failed to execute group request created notification callback") - return - _record_sent("notifications.group_request_created.sent") - return result - - -@hookimpl(wrapper=True) -def access_group_request_completed( - group_request: GroupRequest, - group: Optional[OktaGroup], - requester: OktaUser, - approvers: list[OktaUser], - notify_requester: bool, -) -> Generator[None, None, None]: - try: - result = yield - except Exception: - # Log and do not raise since notification failures should not - # break the flow. Users can still manually ping approvers - # to process their request from the UI - logger.exception("Failed to execute group request completed notification callback") + ``kwargs`` are forwarded verbatim; pluggy passes each implementation only the + parameters it declares, so extra kwargs (e.g. ``requester_role``) are ignored + safely. + """ + relay = get_notification_hook() + _, exceptions = await run_hooks_to_completion( + getattr(relay, hook)(**kwargs), context=f"{hook} notification callback" + ) + if exceptions: + # Failures are already logged; don't record a "sent" for a partial fire. return - _record_sent("notifications.group_request_completed.sent") - return result + metric, tags = _SENT_METRICS[hook] + await _record_sent(metric, tags) def get_notification_hook() -> pluggy.HookRelay: @@ -332,11 +195,9 @@ def get_notification_hook() -> pluggy.HookRelay: pm = pluggy.PluginManager(notification_plugin_name) pm.add_hookspecs(NotificationPluginSpec) - # Register the hook wrappers - pm.register(sys.modules[__name__]) - count = pm.load_setuptools_entrypoints(notification_plugin_name) print(f"Count of loaded notification plugins: {count}") + verify_async_impls(pm, tuple(NotificationHook)) _cached_notification_hook = pm.hook return _cached_notification_hook diff --git a/api/plugins/setup.py b/api/plugins/setup.py index 44aa55d1..f43487d3 100644 --- a/api/plugins/setup.py +++ b/api/plugins/setup.py @@ -2,6 +2,6 @@ setup( name="access-plugins", - install_requires=["pluggy==1.5.0"], + install_requires=["pluggy==1.6.0"], packages=find_packages(), ) diff --git a/api/syncer.py b/api/syncer.py index 8b6cf175..4e2d916d 100644 --- a/api/syncer.py +++ b/api/syncer.py @@ -1,7 +1,6 @@ import logging from collections import defaultdict from datetime import date, datetime, timedelta, timezone -from typing import List from sqlalchemy import and_, func, or_, select from api.config import settings @@ -33,7 +32,7 @@ RejectAccessRequest, UnmanageGroup, ) -from api.plugins import get_notification_hook +from api.plugins import NotificationHook, send_notification from api.services import okta from api.services.okta_service import OktaTimeout, is_managed_group @@ -441,7 +440,6 @@ async def expire_access_requests() -> None: async def expiring_access_notifications_user() -> None: logger.info("Expiring access notifications for users started.") - notification_hook = get_notification_hook() weekend_notif_tomorrow = False day = date.today() + timedelta(days=1) @@ -486,11 +484,6 @@ async def expiring_access_notifications_user() -> None: for membership in db_memberships_expiring_tomorrow: grouped_tomorrow.setdefault(membership.active_user, []).append(membership) - # TODO eventually clean this up, leaving for now for backwards compatibility - grouped_tomorrow_old: dict[OktaUser, list[OktaGroup]] = {} - for membership in db_memberships_expiring_tomorrow: - grouped_tomorrow_old.setdefault(membership.active_user, []).append(membership.active_group) - weekend_notif_week = False day = date.today() + timedelta(weeks=1) next_day = day + timedelta(days=1) @@ -522,65 +515,40 @@ async def expiring_access_notifications_user() -> None: for membership in db_memberships_expiring_next_week: grouped_next_week.setdefault(membership.active_user, []).append(membership) - # TODO eventually clean this up, leaving for now for backwards compatibility - grouped_next_week_old: dict[OktaUser, list[OktaGroup]] = {} - for membership in db_memberships_expiring_next_week: - grouped_next_week_old.setdefault(membership.active_user, []).append(membership.active_group) - for user in grouped_tomorrow: # If the user has access expiring both tomorrow and in a week, only send one message if user in grouped_next_week: - # Notification hooks are sync plugin code holding ORM objects bound to - # this AsyncSession. Run them via run_sync (on the session's own greenlet) - # rather than a worker thread, so those objects are never touched across a - # thread boundary. The syncer is a batch CLI job, so briefly blocking its - # loop while a hook does network I/O is fine (TODO 18: async plugin hooks). - await db.session.run_sync( - lambda _s: notification_hook.access_expiring_user( - # TODO eventually clean this up, leaving for now for backwards compatibility - groups=grouped_tomorrow_old[user] + grouped_next_week_old[user], - user=user, - expiration_datetime=None, - okta_user_group_members=grouped_tomorrow[user] + grouped_next_week[user], - ) + # Notification hooks are native async now: awaited directly on the event + # loop, so the ORM objects they read stay on this AsyncSession without any + # run_sync/worker-thread bridge. + await send_notification( + NotificationHook.ACCESS_EXPIRING_USER, + user=user, + expiration_datetime=None, + okta_user_group_members=grouped_tomorrow[user] + grouped_next_week[user], ) else: - await db.session.run_sync( - lambda _s: notification_hook.access_expiring_user( - # TODO eventually clean this up, leaving for now for backwards compatibility - groups=grouped_tomorrow_old[user], - user=user, - expiration_datetime=None if weekend_notif_tomorrow else datetime.now() + timedelta(days=1), - okta_user_group_members=grouped_tomorrow[user], - ) + await send_notification( + NotificationHook.ACCESS_EXPIRING_USER, + user=user, + expiration_datetime=None if weekend_notif_tomorrow else datetime.now() + timedelta(days=1), + okta_user_group_members=grouped_tomorrow[user], ) for user in grouped_next_week: if user not in grouped_tomorrow: - await db.session.run_sync( - lambda _s: notification_hook.access_expiring_user( - # TODO eventually clean this up, leaving for now for backwards compatibility - groups=grouped_next_week_old[user], - user=user, - expiration_datetime=None if weekend_notif_week else datetime.now() + timedelta(weeks=1), - okta_user_group_members=grouped_next_week[user], - ) + await send_notification( + NotificationHook.ACCESS_EXPIRING_USER, + user=user, + expiration_datetime=None if weekend_notif_week else datetime.now() + timedelta(weeks=1), + okta_user_group_members=grouped_next_week[user], ) logger.info("Expiring access notifications for users finished.") -# TODO eventually clean this up, leaving for now for backwards compatibility -class GroupsAndUsers: - def __init__(self) -> None: - self.groups: List[OktaGroup] = [] - self.roles: List[RoleGroup] = [] - self.users: List[OktaUser] = [] - - async def expiring_access_notifications_owner() -> None: logger.info("Expiring access notifications for owners started.") - notification_hook = get_notification_hook() day = date.today() next_week = day + timedelta(weeks=1) @@ -628,30 +596,6 @@ async def expiring_access_notifications_owner() -> None: if owner.id != okta_user_group_member.user_id: owner_expiring_groups_this[owner].append(okta_user_group_member) - # TODO eventually clean this up, leaving for now for backwards compatibility - # Map of group -> list of users with access expiring next week - users_per_group = defaultdict(list) - - for m in db_memberships_expiring_this_week: - users_per_group[m.active_group].append(m.active_user) - - # Map of group owners -> (number of groups with expiring memberships, number of users with expiring memberships) - owner_expiring_groups_this_old: defaultdict[OktaUser, GroupsAndUsers] = defaultdict(GroupsAndUsers) - for group in users_per_group: - owners = await get_group_managers(group.id) - - if len(owners) == 0: - owners += (await get_app_managers(group.app_id)) if type(group) is AppGroup else [] - - if len(owners) == 0: - owners = access_owners - - for owner in owners: - non_owner_users = [user for user in users_per_group[group] if user.id != owner.id] - if len(non_owner_users) > 0: - owner_expiring_groups_this_old[owner].users += non_owner_users - owner_expiring_groups_this_old[owner].groups.append(group) - one_week = date.today() + timedelta(weeks=1) two_weeks = one_week + timedelta(weeks=1) @@ -690,80 +634,36 @@ async def expiring_access_notifications_owner() -> None: if owner.id != okta_user_group_member.user_id: owner_expiring_groups_next[owner].append(okta_user_group_member) - # TODO eventually clean this up, leaving for now for backwards compatibility - # Map of group -> list of users with access expiring next week - users_per_group = defaultdict(list) - - for m in db_memberships_expiring_next_week: - users_per_group[m.active_group].append(m.active_user) - - # Map of group owners -> (number of groups with expiring memberships, number of users with expiring memberships) - owner_expiring_groups_next_old: defaultdict[OktaUser, GroupsAndUsers] = defaultdict(GroupsAndUsers) - for group in users_per_group: - owners = await get_group_managers(group.id) - - if len(owners) == 0: - owners += (await get_app_managers(group.app_id)) if type(group) is AppGroup else [] - - if len(owners) == 0: - owners = access_owners - - for owner in owners: - non_owner_users = [user for user in users_per_group[group] if user.id != owner.id] - if len(non_owner_users) > 0: - owner_expiring_groups_next_old[owner].users += non_owner_users - owner_expiring_groups_next_old[owner].groups.append(group) - for owner in owner_expiring_groups_this: # If the owner has members with access expiring both this week and next week, only send one message if owner in owner_expiring_groups_next: - # Notification hooks are sync plugin code holding ORM objects bound to - # this AsyncSession. Run them via run_sync (on the session's own greenlet) - # rather than a worker thread, so those objects are never touched across a - # thread boundary. The syncer is a batch CLI job, so briefly blocking its - # loop while a hook does network I/O is fine (TODO 18: async plugin hooks). - await db.session.run_sync( - lambda _s: notification_hook.access_expiring_owner( - owner=owner, - # TODO eventually clean this up, leaving for now for backwards compatibility - groups=owner_expiring_groups_this_old[owner].groups + owner_expiring_groups_next_old[owner].groups, - roles=None, - # TODO eventually clean this up, leaving for now for backwards compatibility - users=owner_expiring_groups_this_old[owner].users + owner_expiring_groups_next_old[owner].users, - expiration_datetime=None, - group_user_associations=owner_expiring_groups_this[owner] + owner_expiring_groups_next[owner], - role_group_associations=None, - ) + # Notification hooks are native async now: awaited directly on the event + # loop, so the ORM objects they read stay on this AsyncSession without any + # run_sync/worker-thread bridge. + await send_notification( + NotificationHook.ACCESS_EXPIRING_OWNER, + owner=owner, + expiration_datetime=None, + group_user_associations=owner_expiring_groups_this[owner] + owner_expiring_groups_next[owner], + role_group_associations=None, ) else: - await db.session.run_sync( - lambda _s: notification_hook.access_expiring_owner( - owner=owner, - # TODO eventually clean this up, leaving for now for backwards compatibility - groups=owner_expiring_groups_this_old[owner].groups, - roles=None, - # TODO eventually clean this up, leaving for now for backwards compatibility - users=owner_expiring_groups_this_old[owner].users, - expiration_datetime=datetime.now(), - group_user_associations=owner_expiring_groups_this[owner], - role_group_associations=None, - ) + await send_notification( + NotificationHook.ACCESS_EXPIRING_OWNER, + owner=owner, + expiration_datetime=datetime.now(), + group_user_associations=owner_expiring_groups_this[owner], + role_group_associations=None, ) for owner in owner_expiring_groups_next: if owner not in owner_expiring_groups_this: - await db.session.run_sync( - lambda _s: notification_hook.access_expiring_owner( - owner=owner, - # TODO eventually clean this up, leaving for now for backwards compatibility - groups=owner_expiring_groups_next_old[owner].groups, - roles=None, - # TODO eventually clean this up, leaving for now for backwards compatibility - users=owner_expiring_groups_next_old[owner].users, - expiration_datetime=datetime.now() + timedelta(weeks=1), - group_user_associations=owner_expiring_groups_next[owner], - role_group_associations=None, - ) + await send_notification( + NotificationHook.ACCESS_EXPIRING_OWNER, + owner=owner, + expiration_datetime=datetime.now() + timedelta(weeks=1), + group_user_associations=owner_expiring_groups_next[owner], + role_group_associations=None, ) role_group_alias = aliased(RoleGroup) @@ -803,28 +703,6 @@ async def expiring_access_notifications_owner() -> None: for owner in owners: owner_expiring_roles_this[owner].append(role_group_map) - # TODO eventually clean this up, leaving for now for backwards compatibility - # Map of group -> list of roles with access expiring next week - roles_per_group = defaultdict(list) - - for r in db_roles_expiring_this_week: - roles_per_group[r.active_group].append(r.active_role_group) - - # Map of group owners -> (number of groups with expiring memberships, number of users with expiring memberships) - owner_expiring_roles_this_old: defaultdict[OktaUser, GroupsAndUsers] = defaultdict(GroupsAndUsers) - for group in roles_per_group: - owners = await get_group_managers(group.id) - - if len(owners) == 0: - owners += (await get_app_managers(group.app_id)) if type(group) is AppGroup else [] - - if len(owners) == 0: - owners = access_owners - - for owner in owners: - owner_expiring_roles_this_old[owner].roles += roles_per_group[group] - owner_expiring_roles_this_old[owner].groups.append(group) - one_week = date.today() + timedelta(weeks=1) two_weeks = one_week + timedelta(weeks=1) @@ -859,73 +737,33 @@ async def expiring_access_notifications_owner() -> None: for owner in owners: owner_expiring_roles_next[owner].append(role_group_map) - # TODO eventually clean this up, leaving for now for backwards compatibility - # Map of group -> list of roles with access expiring next week - roles_per_group = defaultdict(list) - - for r in db_roles_expiring_next_week: - roles_per_group[r.active_group].append(r.active_role_group) - - # Map of group owners -> (number of groups with expiring memberships, number of users with expiring memberships) - owner_expiring_roles_next_old: defaultdict[OktaUser, GroupsAndUsers] = defaultdict(GroupsAndUsers) - for group in roles_per_group: - owners = await get_group_managers(group.id) - - if len(owners) == 0: - owners += (await get_app_managers(group.app_id)) if type(group) is AppGroup else [] - - if len(owners) == 0: - owners = access_owners - - for owner in owners: - owner_expiring_roles_next_old[owner].roles += roles_per_group[group] - owner_expiring_roles_next_old[owner].groups.append(group) - for owner in owner_expiring_roles_this: # If the owner has members with access expiring both this week and next week, only send one message if owner in owner_expiring_roles_next: - await db.session.run_sync( - lambda _s: notification_hook.access_expiring_owner( - owner=owner, - # TODO eventually clean this up, leaving for now for backwards compatibility - groups=owner_expiring_roles_this_old[owner].groups + owner_expiring_roles_next_old[owner].groups, - # TODO eventually clean this up, leaving for now for backwards compatibility - roles=owner_expiring_roles_this_old[owner].roles + owner_expiring_roles_next_old[owner].roles, - users=None, - expiration_datetime=None, - group_user_associations=None, - role_group_associations=owner_expiring_roles_this[owner] + owner_expiring_roles_next[owner], - ) + await send_notification( + NotificationHook.ACCESS_EXPIRING_OWNER, + owner=owner, + expiration_datetime=None, + group_user_associations=None, + role_group_associations=owner_expiring_roles_this[owner] + owner_expiring_roles_next[owner], ) else: - await db.session.run_sync( - lambda _s: notification_hook.access_expiring_owner( - owner=owner, - # TODO eventually clean this up, leaving for now for backwards compatibility - groups=owner_expiring_roles_this_old[owner].groups, - # TODO eventually clean this up, leaving for now for backwards compatibility - roles=owner_expiring_roles_this_old[owner].roles, - users=None, - expiration_datetime=datetime.now(), - group_user_associations=None, - role_group_associations=owner_expiring_roles_this[owner], - ) + await send_notification( + NotificationHook.ACCESS_EXPIRING_OWNER, + owner=owner, + expiration_datetime=datetime.now(), + group_user_associations=None, + role_group_associations=owner_expiring_roles_this[owner], ) for owner in owner_expiring_roles_next: if owner not in owner_expiring_roles_this: - await db.session.run_sync( - lambda _s: notification_hook.access_expiring_owner( - owner=owner, - # TODO eventually clean this up, leaving for now for backwards compatibility - groups=owner_expiring_roles_next_old[owner].groups, - # TODO eventually clean this up, leaving for now for backwards compatibility - roles=owner_expiring_roles_next_old[owner].roles, - users=None, - expiration_datetime=datetime.now() + timedelta(weeks=1), - group_user_associations=None, - role_group_associations=owner_expiring_roles_next[owner], - ) + await send_notification( + NotificationHook.ACCESS_EXPIRING_OWNER, + owner=owner, + expiration_datetime=datetime.now() + timedelta(weeks=1), + group_user_associations=None, + role_group_associations=owner_expiring_roles_next[owner], ) logger.info("Expiring access notifications for owners finished.") @@ -933,7 +771,6 @@ async def expiring_access_notifications_owner() -> None: async def expiring_access_notifications_role_owner() -> None: logger.info("Expiring access notifications for role owners started.") - notification_hook = get_notification_hook() access_owners = await get_access_owners() @@ -1009,35 +846,30 @@ async def expiring_access_notifications_role_owner() -> None: for owner in role_owner_expiring_roles_tomorrow: # If the role owner has roles they own with access expiring both this week and next week, only send one message if owner in role_owner_expiring_roles_next: - # Notification hooks are sync plugin code holding ORM objects bound to - # this AsyncSession. Run them via run_sync (on the session's own greenlet) - # rather than a worker thread, so those objects are never touched across a - # thread boundary. The syncer is a batch CLI job, so briefly blocking its - # loop while a hook does network I/O is fine (TODO 18: async plugin hooks). - await db.session.run_sync( - lambda _s: notification_hook.access_expiring_role_owner( - owner=owner, - roles=role_owner_expiring_roles_tomorrow[owner] + role_owner_expiring_roles_next[owner], - expiration_datetime=None, - ) + # Notification hooks are native async now: awaited directly on the event + # loop, so the ORM objects they read stay on this AsyncSession without any + # run_sync/worker-thread bridge. + await send_notification( + NotificationHook.ACCESS_EXPIRING_ROLE_OWNER, + owner=owner, + roles=role_owner_expiring_roles_tomorrow[owner] + role_owner_expiring_roles_next[owner], + expiration_datetime=None, ) else: - await db.session.run_sync( - lambda _s: notification_hook.access_expiring_role_owner( - owner=owner, - roles=role_owner_expiring_roles_tomorrow[owner], - expiration_datetime=None if weekend_notif_tomorrow else datetime.now() + timedelta(days=1), - ) + await send_notification( + NotificationHook.ACCESS_EXPIRING_ROLE_OWNER, + owner=owner, + roles=role_owner_expiring_roles_tomorrow[owner], + expiration_datetime=None if weekend_notif_tomorrow else datetime.now() + timedelta(days=1), ) for owner in role_owner_expiring_roles_next: if owner not in role_owner_expiring_roles_tomorrow: - await db.session.run_sync( - lambda _s: notification_hook.access_expiring_role_owner( - owner=owner, - roles=role_owner_expiring_roles_next[owner], - expiration_datetime=None if weekend_notif_week else datetime.now() + timedelta(weeks=1), - ) + await send_notification( + NotificationHook.ACCESS_EXPIRING_ROLE_OWNER, + owner=owner, + roles=role_owner_expiring_roles_next[owner], + expiration_datetime=None if weekend_notif_week else datetime.now() + timedelta(weeks=1), ) logger.info("Expiring access notifications for role owners finished.") diff --git a/examples/plugins/app_group_lifecycle_audit_logger/plugin.py b/examples/plugins/app_group_lifecycle_audit_logger/plugin.py index f38c1e6d..4378c637 100644 --- a/examples/plugins/app_group_lifecycle_audit_logger/plugin.py +++ b/examples/plugins/app_group_lifecycle_audit_logger/plugin.py @@ -9,7 +9,7 @@ from datetime import datetime from typing import Any -from sqlalchemy.orm import Session +from sqlalchemy.ext.asyncio import AsyncSession # Import models from api.models import App, AppGroup, OktaUser @@ -191,7 +191,7 @@ def get_plugin_group_status_properties( # Lifecycle hooks @hookimpl - def group_created(self, session: Session, group: AppGroup, plugin_id: str | None) -> None: + async def group_created(self, session: AsyncSession, group: AppGroup, plugin_id: str | None) -> None: """Handle group creation.""" if plugin_id is not None and plugin_id != PLUGIN_ID: return @@ -205,8 +205,8 @@ def group_created(self, session: Session, group: AppGroup, plugin_id: str | None self._increment_event_count(session, group) @hookimpl - def group_updated( - self, session: Session, group: AppGroup, old_name: str, old_description: str, plugin_id: str | None + async def group_updated( + self, session: AsyncSession, group: AppGroup, old_name: str, old_description: str, plugin_id: str | None ) -> None: """Handle group update.""" if plugin_id is not None and plugin_id != PLUGIN_ID: @@ -221,7 +221,7 @@ def group_updated( self._increment_event_count(session, group) @hookimpl - def group_deleted(self, session: Session, group: AppGroup, plugin_id: str | None) -> None: + async def group_deleted(self, session: AsyncSession, group: AppGroup, plugin_id: str | None) -> None: """Handle group deletion.""" if plugin_id is not None and plugin_id != PLUGIN_ID: return @@ -235,9 +235,9 @@ def group_deleted(self, session: Session, group: AppGroup, plugin_id: str | None self._increment_event_count(session, group) @hookimpl - def group_members_added( + async def group_members_added( self, - session: Session, + session: AsyncSession, group: AppGroup, members: list[OktaUser], plugin_id: str | None, @@ -256,9 +256,9 @@ def group_members_added( self._increment_event_count(session, group) @hookimpl - def group_members_removed( + async def group_members_removed( self, - session: Session, + session: AsyncSession, group: AppGroup, members: list[OktaUser], plugin_id: str | None, @@ -277,7 +277,7 @@ def group_members_removed( self._increment_event_count(session, group) @hookimpl - def sync_all_group_membership(self, session: Session, app: App, plugin_id: str | None) -> None: + async def sync_all_group_membership(self, session: AsyncSession, app: App, plugin_id: str | None) -> None: """Perform periodic sync of all group memberships.""" if plugin_id is not None and plugin_id != PLUGIN_ID: return @@ -309,8 +309,13 @@ def _is_enabled(self, group: AppGroup) -> bool: group, "enabled", PLUGIN_ID, True ) - def _increment_event_count(self, session: Session, group: AppGroup) -> None: - """Increment the event count in the status.""" + def _increment_event_count(self, session: AsyncSession, group: AppGroup) -> None: + """Increment the event count in the status. + + Only synchronous work: ``session.add`` is not a coroutine on + ``AsyncSession``, and the get/set status helpers mutate ``plugin_data`` + in memory, so this stays a plain method the async hooks call directly. + """ # Update group-level status current_count = get_status_value(group, "events_logged", PLUGIN_ID) or 0 set_status_value(group, "events_logged", current_count + 1, PLUGIN_ID) diff --git a/examples/plugins/conditional_access/conditional_access.py b/examples/plugins/conditional_access/conditional_access.py index 522d4da6..d51b736d 100644 --- a/examples/plugins/conditional_access/conditional_access.py +++ b/examples/plugins/conditional_access/conditional_access.py @@ -25,7 +25,7 @@ @request_hook_impl -def access_request_created( +async def access_request_created( access_request: AccessRequest, group: OktaGroup, group_tags: List[Tag], requester: OktaUser ) -> Optional[ConditionalAccessResponse]: """Auto-approve memberships to the Auto-Approved-Group group""" diff --git a/examples/plugins/conditional_access/requirements.txt b/examples/plugins/conditional_access/requirements.txt index d43674e1..72806774 100644 --- a/examples/plugins/conditional_access/requirements.txt +++ b/examples/plugins/conditional_access/requirements.txt @@ -1 +1 @@ -pluggy==1.5.0 +pluggy==1.6.0 diff --git a/examples/plugins/datadog_metrics_reporter/metrics_reporter.py b/examples/plugins/datadog_metrics_reporter/metrics_reporter.py index 4e98dccc..ab497382 100644 --- a/examples/plugins/datadog_metrics_reporter/metrics_reporter.py +++ b/examples/plugins/datadog_metrics_reporter/metrics_reporter.py @@ -1,8 +1,8 @@ import logging import os import threading -from contextlib import contextmanager -from typing import Any, ContextManager, Dict, Iterator, List, Optional +from contextlib import asynccontextmanager +from typing import Any, AsyncContextManager, AsyncIterator, Dict, List, Optional import pluggy @@ -76,7 +76,7 @@ def _should_buffer(self) -> bool: return self.batch_depth > 0 @metrics_reporter_hookimpl - def record_counter( + async def record_counter( self, metric_name: str, value: float, tags: Optional[Dict[str, str]], monotonic: bool = True ) -> None: if not self.client: @@ -99,7 +99,7 @@ def record_counter( self.client.increment(metric_name, value=value, tags=formatted_tags) @metrics_reporter_hookimpl - def record_gauge(self, metric_name: str, value: float, tags: Optional[Dict[str, str]]) -> None: + async def record_gauge(self, metric_name: str, value: float, tags: Optional[Dict[str, str]]) -> None: if not self.client: return @@ -114,7 +114,7 @@ def record_gauge(self, metric_name: str, value: float, tags: Optional[Dict[str, self.client.gauge(metric_name, value=value, tags=formatted_tags) @metrics_reporter_hookimpl - def record_histogram( + async def record_histogram( self, metric_name: str, value: float, @@ -140,7 +140,7 @@ def record_histogram( self.client.histogram(metric_name, value=value, tags=formatted_tags) @metrics_reporter_hookimpl - def record_summary(self, metric_name: str, value: float, tags: Optional[Dict[str, str]]) -> None: + async def record_summary(self, metric_name: str, value: float, tags: Optional[Dict[str, str]]) -> None: if not self.client: return @@ -156,9 +156,9 @@ def record_summary(self, metric_name: str, value: float, tags: Optional[Dict[str self.client.distribution(metric_name, value=value, tags=formatted_tags) @metrics_reporter_hookimpl - def batch_metrics(self) -> ContextManager[None]: - @contextmanager - def _batch_context() -> Iterator[None]: + def batch_metrics(self) -> AsyncContextManager[None]: + @asynccontextmanager + async def _batch_context() -> AsyncIterator[None]: with self.batch_lock: self.batch_depth += 1 try: @@ -193,12 +193,12 @@ def _flush_batch(self) -> None: self.batch_buffer.clear() @metrics_reporter_hookimpl - def set_global_tags(self, tags: Dict[str, str]) -> None: + async def set_global_tags(self, tags: Dict[str, str]) -> None: """Set global tags to be included with all metrics.""" self.global_tags.update(tags) @metrics_reporter_hookimpl - def flush(self) -> None: + async def flush(self) -> None: """Force flush any buffered metrics.""" if not self.client: return diff --git a/examples/plugins/datadog_metrics_reporter/requirements.txt b/examples/plugins/datadog_metrics_reporter/requirements.txt index 77cfe33e..ade6cffe 100644 --- a/examples/plugins/datadog_metrics_reporter/requirements.txt +++ b/examples/plugins/datadog_metrics_reporter/requirements.txt @@ -1,2 +1,2 @@ -pluggy==1.5.0 +pluggy==1.6.0 datadog==0.49.0 diff --git a/examples/plugins/notifications/notifications.py b/examples/plugins/notifications/notifications.py index c35aa26d..5bf8be1b 100644 --- a/examples/plugins/notifications/notifications.py +++ b/examples/plugins/notifications/notifications.py @@ -126,7 +126,7 @@ def expiring_access_list_role_owner(expiring_access_list: list[RoleGroupMap]) -> @notification_hook_impl -def access_request_created( +async def access_request_created( access_request: AccessRequest, group: OktaGroup, requester: OktaUser, approvers: list[OktaUser] ) -> None: """Notify all the approvers of the access request through a notification""" @@ -145,7 +145,7 @@ def access_request_created( @notification_hook_impl -def access_request_completed( +async def access_request_completed( access_request: AccessRequest, group: OktaGroup, requester: OktaUser, @@ -165,7 +165,7 @@ def access_request_completed( @notification_hook_impl -def access_role_request_created( +async def access_role_request_created( role_request: RoleRequest, role: RoleGroup, group: OktaGroup, requester: OktaUser, approvers: list[OktaUser] ) -> None: """Notify all the approvers of the role request through a direct message.""" @@ -183,8 +183,7 @@ def access_role_request_created( @notification_hook_impl -def access_expiring_user( - groups: list[OktaGroup], +async def access_expiring_user( user: OktaUser, expiration_datetime: datetime, okta_user_group_members: list[OktaUserGroupMember], @@ -205,11 +204,8 @@ def access_expiring_user( @notification_hook_impl -def access_expiring_owner( +async def access_expiring_owner( owner: OktaUser, - groups: list[OktaGroup], - roles: list[RoleGroup], - users: list[OktaUser], expiration_datetime: datetime, group_user_associations: Optional[list[OktaUserGroupMember]], role_group_associations: Optional[list[RoleGroupMap]], @@ -254,7 +250,7 @@ def access_expiring_owner( @notification_hook_impl -def access_expiring_role_owner(owner: OktaUser, roles: list[RoleGroupMap], expiration_datetime: datetime) -> None: +async def access_expiring_role_owner(owner: OktaUser, roles: list[RoleGroupMap], expiration_datetime: datetime) -> None: expiring_access_url = get_base_url() + "/expiring-roles?role_owner_id=@me" expiring_access_list = expiring_access_list_role_owner(roles) diff --git a/examples/plugins/notifications/requirements.txt b/examples/plugins/notifications/requirements.txt index d43674e1..72806774 100644 --- a/examples/plugins/notifications/requirements.txt +++ b/examples/plugins/notifications/requirements.txt @@ -1 +1 @@ -pluggy==1.5.0 +pluggy==1.6.0 diff --git a/examples/plugins/notifications_slack/notifications.py b/examples/plugins/notifications_slack/notifications.py index c609ea85..22ebe2f0 100644 --- a/examples/plugins/notifications_slack/notifications.py +++ b/examples/plugins/notifications_slack/notifications.py @@ -1,14 +1,14 @@ from __future__ import print_function +import asyncio import logging import os import random -import time from datetime import date, datetime, timedelta -from typing import Any, Callable, Dict, List, Optional, TypeVar +from typing import Any, Awaitable, Callable, Dict, List, Optional, TypeVar import pluggy -from slack_sdk import WebClient +from slack_sdk.web.async_client import AsyncWebClient from api.models import ( AccessRequest, @@ -28,14 +28,14 @@ T = TypeVar("T") -def retry_operation( - operation_func: Callable[[], T], max_attempts: int = 3, base_delay: float = 1.0, max_delay: float = 10.0 +async def retry_operation( + operation_func: Callable[[], Awaitable[T]], max_attempts: int = 3, base_delay: float = 1.0, max_delay: float = 10.0 ) -> Optional[T]: """ - Execute an operation with retries, without propagating exceptions + Await an async operation with retries, without propagating exceptions. Args: - operation_func: Function that performs the operation + operation_func: Async function that performs the operation max_attempts: Maximum number of retry attempts base_delay: Initial delay between retries in seconds max_delay: Maximum delay between retries in seconds @@ -48,7 +48,7 @@ def retry_operation( while attempt < max_attempts: try: - return operation_func() + return await operation_func() except Exception as e: attempt += 1 @@ -63,15 +63,16 @@ def retry_operation( logger.warning( f"Operation failed (attempt {attempt}/{max_attempts}): {str(e)}. Retrying in {sleep_time:.2f}s" ) - time.sleep(sleep_time) + await asyncio.sleep(sleep_time) delay = min(delay * 2, max_delay) # Exponential backoff return None -# Initialize Slack client and signature verifier +# Initialize Slack client. AsyncWebClient is the Slack SDK's native-async client +# (aiohttp-backed), so its calls are awaited directly on the event loop. slack_token = os.environ["SLACK_BOT_TOKEN"] -client = WebClient(token=slack_token) +client = AsyncWebClient(token=slack_token) alerts_channel = os.environ.get("SLACK_ALERTS_CHANNEL") CLIENT_ORIGIN_URL = os.environ.get("CLIENT_ORIGIN_URL") # e.g. "https://discord-access-instance.com" @@ -238,7 +239,7 @@ def expiring_access_list_role_owner(expiring_access_list: List[RoleGroupMap]) -> return out -def get_user_id_by_email(email: str) -> Optional[str]: +async def get_user_id_by_email(email: str) -> Optional[str]: """Get Slack user ID by email with retry logic. Args: @@ -248,67 +249,74 @@ def get_user_id_by_email(email: str) -> Optional[str]: Optional[str]: The Slack user ID if found, otherwise None. """ - def lookup_user() -> str: - response = client.users_lookupByEmail(email=email) + async def lookup_user() -> str: + response = await client.users_lookupByEmail(email=email) return response["user"]["id"] - user_id = retry_operation(lookup_user) + user_id = await retry_operation(lookup_user) if not user_id: logger.error(f"Failed to fetch user ID for {email} after multiple attempts") return user_id -def send_slack_dm(user: OktaUser, message: str) -> None: +async def send_slack_dm(user: OktaUser, message: str) -> None: """Send a direct message to a Slack user with retry logic. + Uses the Slack SDK's native-async ``AsyncWebClient``, so the API calls are + awaited directly on the event loop — no thread offload needed (an async + client, awaited on the event loop). + Args: user (OktaUser): The user to send the message to. message (str): The message content. """ - user_id = get_user_id_by_email(user.email) + user_id = await get_user_id_by_email(user.email) if user_id: mention_message = f"<@{user_id}> {message}" - def send_message() -> Dict[str, Any]: - response = client.chat_postMessage( + async def send_message() -> Dict[str, Any]: + response = await client.chat_postMessage( channel=user_id, text=mention_message, as_user=True, unfurl_links=True, unfurl_media=True ) logger.info(f"Slack DM sent: {response['ts']}") return response - result = retry_operation(send_message) + result = await retry_operation(send_message) if not result: logger.error(f"Failed to send Slack DM to {user.email} after multiple attempts") -def send_slack_channel_message(user: OktaUser, message: str) -> None: +async def send_slack_channel_message(user: OktaUser, message: str) -> None: """Send a message to a Slack channel with retry logic. + See ``send_slack_dm`` — the Slack calls are awaited on the async client. + Args: message (str): The message content. user (OktaUser): The user to relate the message to. """ - if alerts_channel: - user_id = get_user_id_by_email(user.email) + if not alerts_channel: + return - if user_id: - channel_message = f"{user.email} - {message}" + user_id = await get_user_id_by_email(user.email) + if user_id: + channel_message = f"{user.email} - {message}" - def send_message() -> Dict[str, Any]: - response = client.chat_postMessage( - channel=alerts_channel, text=channel_message, as_user=True, unfurl_links=True, unfurl_media=True - ) - logger.info(f"Slack channel message sent: {response['ts']}") - return response + async def send_message() -> Dict[str, Any]: + response = await client.chat_postMessage( + channel=alerts_channel, text=channel_message, as_user=True, unfurl_links=True, unfurl_media=True + ) + logger.info(f"Slack channel message sent: {response['ts']}") + return response - result = retry_operation(send_message) - if not result: - logger.error(f"Failed to send message to channel {alerts_channel} after multiple attempts") + result = await retry_operation(send_message) + if not result: + logger.error(f"Failed to send message to channel {alerts_channel} after multiple attempts") @notification_hook_impl -def access_request_created( +async def access_request_created( access_request: AccessRequest, group: OktaGroup, requester: OktaUser, approvers: List[OktaUser] ) -> None: """Notify all the approvers of the access request through a notification. @@ -332,20 +340,20 @@ def access_request_created( # Send the message to the approvers for approver in approvers: - send_slack_dm(approver, approver_message) + await send_slack_dm(approver, approver_message) logger.info(f"Approver message: {approver_message}") # Send the message to the requester only if they're not already an approver if requester.id not in [approver.id for approver in approvers]: - send_slack_dm(requester, approver_message) + await send_slack_dm(requester, approver_message) logger.info("Requester received creation notification") # Post to the alerts channel - send_slack_channel_message(requester, approver_message) + await send_slack_channel_message(requester, approver_message) @notification_hook_impl -def access_role_request_created( +async def access_role_request_created( role_request: RoleRequest, role: RoleGroup, group: OktaGroup, requester: OktaUser, approvers: List[OktaUser] ) -> None: """Notify all the approvers of the role request through a notification. @@ -368,20 +376,22 @@ def access_role_request_created( # Send the message to the approvers for approver in approvers: - send_slack_dm(approver, approver_message) + await send_slack_dm(approver, approver_message) logger.info(f"Approver message: {approver_message}") # Send the message to the requester only if they're not already an approver if requester.id not in [approver.id for approver in approvers]: - send_slack_dm(requester, approver_message) + await send_slack_dm(requester, approver_message) logger.info("Requester received creation notification") # Post to the alerts channel - send_slack_channel_message(requester, approver_message) + await send_slack_channel_message(requester, approver_message) @notification_hook_impl -def access_group_request_created(group_request: GroupRequest, requester: OktaUser, approvers: List[OktaUser]) -> None: +async def access_group_request_created( + group_request: GroupRequest, requester: OktaUser, approvers: List[OktaUser] +) -> None: """Notify the approvers that a group creation has been requested. Args: @@ -406,20 +416,20 @@ def access_group_request_created(group_request: GroupRequest, requester: OktaUse # Send the message to the approvers for approver in approvers: - send_slack_dm(approver, approver_message) + await send_slack_dm(approver, approver_message) logger.info(f"Approver message: {approver_message}") # Send the message to the requester only if they're not already an approver if requester.id not in [approver.id for approver in approvers]: - send_slack_dm(requester, approver_message) + await send_slack_dm(requester, approver_message) logger.info("Requester received group request creation notification") # Post to the alerts channel - send_slack_channel_message(requester, approver_message) + await send_slack_channel_message(requester, approver_message) @notification_hook_impl -def access_request_completed( +async def access_request_completed( access_request: AccessRequest, group: OktaGroup, requester: OktaUser, @@ -442,21 +452,21 @@ def access_request_completed( ) # Send the message to the requester - send_slack_dm(requester, requester_message) + await send_slack_dm(requester, requester_message) logger.info(f"Requester message: {requester_message}") # Send the message to all approvers (except the requester) for approver in approvers: if approver.id != requester.id: # Skip if approver is the requester - send_slack_dm(approver, requester_message) + await send_slack_dm(approver, requester_message) logger.info("Approvers received completion notification") # Post to the alerts channel - send_slack_channel_message(requester, requester_message) + await send_slack_channel_message(requester, requester_message) @notification_hook_impl -def access_group_request_completed( +async def access_group_request_completed( group_request: GroupRequest, group: Optional[OktaGroup], requester: OktaUser, @@ -486,22 +496,21 @@ def access_group_request_completed( # Send the message to the requester if notify_requester: - send_slack_dm(requester, message) + await send_slack_dm(requester, message) logger.info(f"Requester message: {message}") # Send the message to all approvers (except the requester) for approver in approvers: if approver.id != requester.id: - send_slack_dm(approver, message) + await send_slack_dm(approver, message) logger.info("Approvers received group request completion notification") # Post to the alerts channel - send_slack_channel_message(requester, message) + await send_slack_channel_message(requester, message) @notification_hook_impl -def access_expiring_user( - groups: List[OktaGroup], +async def access_expiring_user( user: OktaUser, expiration_datetime: datetime, okta_user_group_members: List[OktaUserGroupMember], @@ -509,7 +518,6 @@ def access_expiring_user( """Notify individuals that their access to a group is expiring soon. Args: - groups (List[OktaGroup]): The list of groups. user (OktaUser): The user whose access is expiring. expiration_datetime (datetime): The expiration date and time. okta_user_group_members (List[OktaUserGroupMember]): List of expiring memberships and ownerships. @@ -526,19 +534,16 @@ def access_expiring_user( ) # Send the message to the individual user with expiring access - send_slack_dm(user, message) + await send_slack_dm(user, message) logger.info(f"User message: {message}") # Post to the alerts channel - send_slack_channel_message(user, message) + await send_slack_channel_message(user, message) @notification_hook_impl -def access_expiring_owner( +async def access_expiring_owner( owner: OktaUser, - groups: List[OktaGroup], - roles: List[RoleGroup], - users: List[OktaUser], expiration_datetime: datetime, group_user_associations: Optional[List[OktaUserGroupMember]], role_group_associations: Optional[List[RoleGroupMap]], @@ -547,9 +552,6 @@ def access_expiring_owner( Args: owner (OktaUser): The owner of the group. - groups (List[OktaGroup]): The list of groups. - roles (List[OktaGroup]): The list of roles. - users (List[RoleGroup]): The list of users. expiration_datetime (datetime): The expiration date and time. group_user_associations (Optional[List[OktaUserGroupMember]]): List of memberships and ownerships expiring. role_group_associations (Optional[List[RoleGroupMap]]): List of role memberships and ownerships expiring. @@ -571,11 +573,11 @@ def access_expiring_owner( ) # Send the message to the group owner about the users with expiring access - send_slack_dm(owner, message) + await send_slack_dm(owner, message) logger.info(f"Owner message: {message}") # Post to the alerts channel - send_slack_channel_message(owner, message) + await send_slack_channel_message(owner, message) if role_group_associations is not None and len(role_group_associations) > 0: expiring_access_url = get_base_url() + "/expiring-roles?owner_id=@me" @@ -594,15 +596,15 @@ def access_expiring_owner( ) # Send the message to the group owner about the roles with expiring access - send_slack_dm(owner, message) + await send_slack_dm(owner, message) logger.info(f"Owner message: {message}") # Post to the alerts channel - send_slack_channel_message(owner, message) + await send_slack_channel_message(owner, message) @notification_hook_impl -def access_expiring_role_owner(owner: OktaUser, roles: List[RoleGroupMap], expiration_datetime: datetime) -> None: +async def access_expiring_role_owner(owner: OktaUser, roles: List[RoleGroupMap], expiration_datetime: datetime) -> None: """Notify role owners that roles they own will be losing access to groups soon. Args: @@ -623,8 +625,8 @@ def access_expiring_role_owner(owner: OktaUser, roles: List[RoleGroupMap], expir ) # Send the message to the group owner about the roles with expiring access - send_slack_dm(owner, message) + await send_slack_dm(owner, message) logger.info(f"Role owner message: {message}") # Post to the alerts channel - send_slack_channel_message(owner, message) + await send_slack_channel_message(owner, message) diff --git a/examples/plugins/notifications_slack/requirements.txt b/examples/plugins/notifications_slack/requirements.txt index e183809f..102869b6 100644 --- a/examples/plugins/notifications_slack/requirements.txt +++ b/examples/plugins/notifications_slack/requirements.txt @@ -1,2 +1,3 @@ -pluggy==1.5.0 +pluggy==1.6.0 slack-sdk==3.27.2 +aiohttp==3.14.1 diff --git a/pyproject.toml b/pyproject.toml index beddb1f1..48c4a0bc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,7 +38,7 @@ dependencies = [ # Okta "okta==2.9.8", # Plugins - "pluggy==1.5.0", + "pluggy==1.6.0", # MCP — embedded Model Context Protocol server (gated behind ENABLE_MCP). # Bumping the major (2.x) will break imports — FastMCP's API has # historically reshaped between majors. diff --git a/tests/test_access_request.py b/tests/test_access_request.py index 2e0b8962..222586d9 100644 --- a/tests/test_access_request.py +++ b/tests/test_access_request.py @@ -28,7 +28,7 @@ ModifyRoleGroups, RejectAccessRequest, ) -from api.plugins import ConditionalAccessResponse, get_conditional_access_hook, get_notification_hook +from api.plugins import ConditionalAccessResponse, get_notification_hook from api.services import okta from tests.factories import AccessRequestFactory, AppGroupFactory, OktaGroupFactory, OktaUserFactory from tests.helpers import db_count @@ -796,10 +796,10 @@ async def test_auto_resolve_create_access_request( notification_hook = get_notification_hook() request_created_notification_spy = mocker.patch.object(notification_hook, "access_request_created") request_completed_notification_spy = mocker.patch.object(notification_hook, "access_request_completed") - request_hook = get_conditional_access_hook() - request_created_conditional_access_spy = mocker.patch.object( - request_hook, - "access_request_created", + # Conditional-access hooks are async; patch the dispatch helper the operation + # awaits (mock.patch autospecs it to an AsyncMock since it is a coroutine fn). + request_created_conditional_access_spy = mocker.patch( + "api.operations.create_access_request.evaluate_conditional_access", return_value=[ConditionalAccessResponse(approved=True, reason="Auto-Approved")], ) add_membership_spy = mocker.patch.object(okta, "add_user_to_group") @@ -835,9 +835,8 @@ async def test_auto_resolve_create_access_request( request_created_conditional_access_spy.reset_mock() add_membership_spy.reset_mock() - request_created_conditional_access_spy = mocker.patch.object( - request_hook, - "access_request_created", + request_created_conditional_access_spy = mocker.patch( + "api.operations.create_access_request.evaluate_conditional_access", return_value=[ConditionalAccessResponse(approved=False, reason="Auto-Rejected")], ) @@ -871,8 +870,8 @@ async def test_auto_resolve_create_access_request( request_created_conditional_access_spy.reset_mock() add_membership_spy.reset_mock() - request_created_conditional_access_spy = mocker.patch.object( - request_hook, "access_request_created", return_value=[None] + request_created_conditional_access_spy = mocker.patch( + "api.operations.create_access_request.evaluate_conditional_access", return_value=[None] ) access_request = await CreateAccessRequest( @@ -918,10 +917,8 @@ async def test_auto_resolve_create_access_request_with_time_limit_constraint_tag notification_hook = get_notification_hook() request_created_notification_spy = mocker.patch.object(notification_hook, "access_request_created") request_completed_notification_spy = mocker.patch.object(notification_hook, "access_request_completed") - request_hook = get_conditional_access_hook() - request_created_conditional_access_spy = mocker.patch.object( - request_hook, - "access_request_created", + request_created_conditional_access_spy = mocker.patch( + "api.operations.create_access_request.evaluate_conditional_access", return_value=[ ConditionalAccessResponse( approved=True, diff --git a/tests/test_app_group_lifecycle_plugin.py b/tests/test_app_group_lifecycle_plugin.py index 0bc5c7f6..1d91aec0 100644 --- a/tests/test_app_group_lifecycle_plugin.py +++ b/tests/test_app_group_lifecycle_plugin.py @@ -16,7 +16,7 @@ from httpx import AsyncClient from pytest_mock import MockerFixture from sqlalchemy import select -from sqlalchemy.orm import Session +from sqlalchemy.ext.asyncio import AsyncSession from api.config import settings from api.extensions import Db @@ -161,14 +161,14 @@ def get_plugin_group_status_properties( } @hookimpl - def group_created(self, session: Session, group: AppGroup, plugin_id: str | None) -> None: + async def group_created(self, session: AsyncSession, group: AppGroup, plugin_id: str | None) -> None: if plugin_id is not None and plugin_id != self.ID: return - # Read through the session synchronously. The hookspec promises a sync - # `Session`; if an operation hands us the raw AsyncSession (a missing - # `run_sync` bridge), this raises and the call below is never recorded — - # so the recorded-calls assertions in these tests fail loudly. - session.scalars(select(AppGroup)).all() + # Await a read through the AsyncSession. The hookspec promises an + # AsyncSession; if an operation hands us something else (a broken bridge), + # this raises and the call below is never recorded — so the recorded-calls + # assertions in these tests fail loudly. + (await session.scalars(select(AppGroup))).all() # A hook may also read group.app, which is lazy="raise_on_sql". The invoking # operation must eager-load AppGroup.app (or seed the identity map) so this # resolves without emitting SQL; otherwise it raises here and the recorded- @@ -177,39 +177,39 @@ def group_created(self, session: Session, group: AppGroup, plugin_id: str | None self.group_created_calls.append(group.id) @hookimpl - def group_updated( - self, session: Session, group: AppGroup, old_name: str, old_description: str, plugin_id: str | None + async def group_updated( + self, session: AsyncSession, group: AppGroup, old_name: str, old_description: str, plugin_id: str | None ) -> None: if plugin_id is not None and plugin_id != self.ID: return - session.scalars(select(AppGroup)).all() # exercise the sync Session (see group_created) + (await session.scalars(select(AppGroup))).all() # exercise the AsyncSession (see group_created) _ = group.app.name # exercise group.app eager-load (see group_created) self.group_updated_calls.append((group.id, old_name, old_description)) @hookimpl - def group_deleted(self, session: Session, group: AppGroup, plugin_id: str | None) -> None: + async def group_deleted(self, session: AsyncSession, group: AppGroup, plugin_id: str | None) -> None: if plugin_id is not None and plugin_id != self.ID: return - session.scalars(select(AppGroup)).all() # exercise the sync Session (see group_created) + (await session.scalars(select(AppGroup))).all() # exercise the AsyncSession (see group_created) _ = group.app.name # exercise group.app eager-load (see group_created) self.group_deleted_calls.append(group.id) @hookimpl - def group_members_added( - self, session: Session, group: AppGroup, members: list[OktaUser], plugin_id: str | None + async def group_members_added( + self, session: AsyncSession, group: AppGroup, members: list[OktaUser], plugin_id: str | None ) -> None: if plugin_id is not None and plugin_id != self.ID: return - session.scalars(select(AppGroup)).all() # exercise the sync Session (see group_created) + (await session.scalars(select(AppGroup))).all() # exercise the AsyncSession (see group_created) self.members_added_calls.append((group.id, [m.id for m in members])) @hookimpl - def group_members_removed( - self, session: Session, group: AppGroup, members: list[OktaUser], plugin_id: str | None + async def group_members_removed( + self, session: AsyncSession, group: AppGroup, members: list[OktaUser], plugin_id: str | None ) -> None: if plugin_id is not None and plugin_id != self.ID: return - session.scalars(select(AppGroup)).all() # exercise the sync Session (see group_created) + (await session.scalars(select(AppGroup))).all() # exercise the AsyncSession (see group_created) self.members_removed_calls.append((group.id, [m.id for m in members])) diff --git a/tests/test_async_dispatch.py b/tests/test_async_dispatch.py new file mode 100644 index 00000000..bf03ccab --- /dev/null +++ b/tests/test_async_dispatch.py @@ -0,0 +1,83 @@ +"""Tests for the async plugin-hook runner (api/plugins/_async_dispatch.py). + +These lock in the ``asyncio.wait`` (not ``gather``) behavior: a cancellation of +the awaiting caller must NOT tear down the in-flight hook coroutines, and one +hook failing must not cancel or abandon its siblings. Both assertions fail if +``run_hooks_to_completion`` is "simplified" back to ``asyncio.gather``. +""" + +from __future__ import annotations + +import asyncio + +import pytest + +from api.plugins._async_dispatch import run_hooks_to_completion + + +async def test_empty_returns_empty() -> None: + results, exceptions = await run_hooks_to_completion([], context="test") + assert results == [] + assert exceptions == [] + + +async def test_collects_results_and_exceptions_in_input_order() -> None: + async def ok_a() -> str: + return "A" + + async def boom() -> str: + raise RuntimeError("kaboom") + + async def ok_b() -> str: + return "B" + + results, exceptions = await run_hooks_to_completion([ok_a(), boom(), ok_b()], context="test") + + assert results == ["A", "B"] + assert len(exceptions) == 1 + assert isinstance(exceptions[0], RuntimeError) + + +async def test_sibling_failure_does_not_cancel_other_inflight_hooks() -> None: + """A hook that raises immediately must not cancel a slow sibling mid-flight.""" + completed: list[str] = [] + + async def boom() -> None: + raise RuntimeError("fails fast") + + async def slow() -> str: + await asyncio.sleep(0.05) + completed.append("slow") # only reached if not cancelled + return "slow" + + results, exceptions = await run_hooks_to_completion([boom(), slow()], context="test") + + assert completed == ["slow"] # ran to completion despite the sibling's failure + assert results == ["slow"] + assert len(exceptions) == 1 + + +async def test_caller_cancellation_does_not_tear_down_inflight_hooks() -> None: + """Cancelling the awaiting task must leave in-flight hook coroutines running. + + This is the whole reason for ``wait`` over ``gather``: ``gather`` would + propagate the cancellation to its children, so ``completed`` would stay empty + and this test would fail. + """ + completed: list[str] = [] + started = asyncio.Event() + + async def slow() -> None: + started.set() + await asyncio.sleep(0.05) + completed.append("done") # only reached if the hook was NOT cancelled + + task = asyncio.ensure_future(run_hooks_to_completion([slow()], context="test")) + await started.wait() # make sure slow() is genuinely in-flight + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + # Give the (uncancelled) hook time to finish on the loop. + await asyncio.sleep(0.2) + assert completed == ["done"] diff --git a/tests/test_expiring_access_notifications.py b/tests/test_expiring_access_notifications.py index 7844110b..40c68053 100644 --- a/tests/test_expiring_access_notifications.py +++ b/tests/test_expiring_access_notifications.py @@ -68,10 +68,6 @@ async def test_individual_expiring_access_notifications( assert kwargs["expiration_datetime"] is None else: assert expiration_datetime.date() == kwargs["expiration_datetime"].date() - # TODO eventually clean this up, leaving for now for backwards compatibility - assert len(kwargs["groups"]) == 2 - assert okta_group in kwargs["groups"] - assert role_group in kwargs["groups"] # Test with one user who has a membership expiring tomorrow and one in a week @@ -120,10 +116,6 @@ async def test_individual_expiring_access_notifications_week( assert membership2 in kwargs["okta_user_group_members"] assert user == kwargs["user"] assert kwargs["expiration_datetime"] is None - # TODO eventually clean this up, leaving for now for backwards compatibility - assert len(kwargs["groups"]) == 2 - assert okta_group in kwargs["groups"] - assert role_group in kwargs["groups"] # Test with one user who has one direct membership expiring tomorrow and a role membership for the same group @@ -237,14 +229,6 @@ async def test_owner_expiring_access_notifications(db: Db, mocker: MockerFixture assert membership1 in kwargs["group_user_associations"] assert membership2 in kwargs["group_user_associations"] assert kwargs["role_group_associations"] is None - # TODO eventually clean this up, leaving for now for backwards compatibility - assert len(kwargs["groups"]) == 2 - assert group1 in kwargs["groups"] - assert group2 in kwargs["groups"] - assert len(kwargs["users"]) == 2 - assert user1 in kwargs["users"] - assert user2 in kwargs["users"] - assert kwargs["roles"] is None # Test with one owner who owns one group, the owner is a member of the group and their access is expiring this week @@ -330,14 +314,6 @@ async def test_owner_expiring_access_notifications_owner_member(db: Db, mocker: assert membership2 not in kwargs["group_user_associations"] assert membership3 in kwargs["group_user_associations"] assert kwargs["role_group_associations"] is None - # TODO eventually clean this up, leaving for now for backwards compatibility - assert len(kwargs["groups"]) == 2 - assert group1 in kwargs["groups"] - assert group2 in kwargs["groups"] - assert len(kwargs["users"]) == 2 - assert user1 in kwargs["users"] - assert user2 in kwargs["users"] - assert kwargs["roles"] is None # Test with one owner who owns two groups, each group has a member whose access expires next week @@ -391,14 +367,6 @@ async def test_owner_expiring_access_notifications_week(db: Db, mocker: MockerFi assert membership1 in kwargs["group_user_associations"] assert membership2 in kwargs["group_user_associations"] assert kwargs["role_group_associations"] is None - # TODO eventually clean this up, leaving for now for backwards compatibility - assert len(kwargs["groups"]) == 2 - assert group1 in kwargs["groups"] - assert group2 in kwargs["groups"] - assert len(kwargs["users"]) == 2 - assert user1 in kwargs["users"] - assert user2 in kwargs["users"] - assert kwargs["roles"] is None # Test with one owner who owns one group, the owner is a member of the group and their access is expiring next week @@ -484,14 +452,6 @@ async def test_owner_expiring_access_notifications_owner_member_week(db: Db, moc assert membership2 not in kwargs["group_user_associations"] assert membership3 in kwargs["group_user_associations"] assert kwargs["role_group_associations"] is None - # TODO eventually clean this up, leaving for now for backwards compatibility - assert len(kwargs["groups"]) == 2 - assert group1 in kwargs["groups"] - assert group2 in kwargs["groups"] - assert len(kwargs["users"]) == 2 - assert user1 in kwargs["users"] - assert user2 in kwargs["users"] - assert kwargs["roles"] is None # Test with one owner who owns a groups, the group has a role member whose access expires this week @@ -533,12 +493,6 @@ async def test_owner_expiring_access_notifications_role(db: Db, mocker: MockerFi _, kwargs = expiring_access_notification_spy.call_args assert len(kwargs["role_group_associations"]) == 1 assert membership1 in kwargs["role_group_associations"] - # TODO eventually clean this up, leaving for now for backwards compatibility - assert len(kwargs["groups"]) == 1 - assert group2 in kwargs["groups"] - assert kwargs["users"] is None - assert len(kwargs["roles"]) == 1 - assert group1 in kwargs["roles"] # Test with one owner who owns two groups, each group has a role member whose access expires next week @@ -590,13 +544,6 @@ async def test_owner_expiring_access_notifications_role_week(db: Db, mocker: Moc assert len(kwargs["role_group_associations"]) == 2 assert membership1 in kwargs["role_group_associations"] assert membership2 in kwargs["role_group_associations"] - # TODO eventually clean this up, leaving for now for backwards compatibility - assert len(kwargs["groups"]) == 2 - assert group1 in kwargs["groups"] - assert group2 in kwargs["groups"] - assert kwargs["users"] is None - assert len(kwargs["roles"]) == 2 - assert role in kwargs["roles"] # Test should not renew funtionality for individual notifications @@ -638,8 +585,6 @@ async def test_individual_do_not_renew_notification_behavior( assert expiring_access_notification_spy.call_count == 1 _, kwargs = expiring_access_notification_spy.call_args - assert len(kwargs["groups"]) == 1 - assert okta_group in kwargs["groups"] assert user == kwargs["user"] if datetime.now().weekday() == 4: assert kwargs["expiration_datetime"] is None @@ -873,19 +818,12 @@ async def test_owner_expiring_access_notifications_managed_group_admin( assert len(kwargs["group_user_associations"]) == 1 assert membership in kwargs["group_user_associations"] assert kwargs["role_group_associations"] is None - # TODO eventually clean this up, leaving for now for backwards compatibility - assert len(kwargs["groups"]) == 1 - assert group2 in kwargs["groups"] - assert len(kwargs["users"]) == 1 - assert user in kwargs["users"] - assert kwargs["roles"] is None - - -# The syncer offloads notification hooks (sync plugin code) while holding ORM objects -# bound to the request's AsyncSession. It invokes them via db.session.run_sync so they -# run on the event loop's own greenlet — never a worker thread — which is what keeps -# passing those session-bound objects safe. This guards that contract: the hook must -# run on the loop thread and must be handed usable, eager-loaded ORM objects. + + +# Notification hooks are now native async: the syncer awaits them directly on the event +# loop while holding ORM objects bound to its AsyncSession — no run_sync/worker-thread +# bridge. This guards that contract: the hook runs on the loop thread and is +# handed usable, eager-loaded ORM objects it can read without emitting SQL. async def test_expiring_user_notification_hook_runs_on_loop_thread_with_usable_orm_objects( db: Db, mocker: MockerFixture, user: OktaUser, okta_group: OktaGroup ) -> None: @@ -903,7 +841,7 @@ async def test_expiring_user_notification_hook_runs_on_loop_thread_with_usable_o hook = get_notification_hook() observed: dict[str, object] = {} - def _record(**kwargs: Any) -> None: + def _record(**kwargs: Any) -> list[Any]: member = kwargs["okta_user_group_members"][0] # The syncer eager-loads active_user/active_group, so a hook can read them # (and their columns) without emitting SQL. @@ -915,10 +853,12 @@ def _record(**kwargs: Any) -> None: # because active_group put the row in the identity map and the many-to-one # resolves from it via use_get without emitting SQL. Guards that eager-load. observed["nonactive_group_name"] = member.group.name - # run_sync executes the hook on the event loop's thread (a greenlet), not an - # anyio worker thread. If this were reverted to anyio.to_thread.run_sync the - # hook would run on a worker thread and this id would differ from the test's. + # The hook is awaited directly on the event loop (no run_sync/worker thread), + # so it runs on the test's own thread. observed["thread_id"] = threading.get_ident() + # send_notification unpacks the caller's result with `*` and awaits it; async + # hookimpls return coroutines, so this stand-in returns an (empty) iterable. + return [] mocker.patch.object(hook, "access_expiring_user", side_effect=_record) @@ -956,11 +896,12 @@ async def test_expiring_owner_notification_hook_can_read_association_relationshi observed: dict[str, object] = {} - def _record(**kwargs: Any) -> None: + def _record(**kwargs: Any) -> list[Any]: assoc = kwargs["group_user_associations"][0] # Non-active .user / .group relationships (not the eager-loaded active_* ones). observed["member_email"] = assoc.user.email observed["group_name"] = assoc.group.name + return [] # send_notification unpacks and awaits the caller's result mocker.patch.object(get_notification_hook(), "access_expiring_owner", side_effect=_record) @@ -989,11 +930,12 @@ async def test_expiring_role_owner_notification_hook_can_read_association_relati observed: dict[str, object] = {} - def _record(**kwargs: Any) -> None: + def _record(**kwargs: Any) -> list[Any]: rgm = kwargs["roles"][0] # Non-active .role_group / .group relationships on the RoleGroupMap row. observed["role_name"] = rgm.role_group.name observed["group_name"] = rgm.group.name + return [] # send_notification unpacks and awaits the caller's result mocker.patch.object(get_notification_hook(), "access_expiring_role_owner", side_effect=_record) diff --git a/tests/test_metrics_reporter_validation.py b/tests/test_metrics_reporter_validation.py index cc5acb7d..1ff35c84 100644 --- a/tests/test_metrics_reporter_validation.py +++ b/tests/test_metrics_reporter_validation.py @@ -5,6 +5,7 @@ import pluggy import pytest +from api.plugins._async_dispatch import verify_async_impls from api.plugins.metrics_reporter import ( MetricsReporterPluginSpec, _verify_tag_forwarding, @@ -41,3 +42,27 @@ def record_counter( with pytest.raises(RuntimeError, match="record_counter"): _verify_tag_forwarding(_manager_with(BadPlugin())) + + +def test_verify_async_impls_rejects_sync_hook() -> None: + class SyncPlugin: + @hookimpl + def record_counter( + self, metric_name: str, value: float, tags: Optional[dict[str, str]], monotonic: bool = True + ) -> None: + pass + + with pytest.raises(RuntimeError, match="async def"): + verify_async_impls(_manager_with(SyncPlugin()), ("record_counter",)) + + +def test_verify_async_impls_accepts_async_hook() -> None: + class AsyncPlugin: + @hookimpl + async def record_counter( + self, metric_name: str, value: float, tags: Optional[dict[str, str]], monotonic: bool = True + ) -> None: + pass + + # Should not raise. + verify_async_impls(_manager_with(AsyncPlugin()), ("record_counter",)) diff --git a/tests/test_notification_metrics.py b/tests/test_notification_metrics.py index 55c44b92..4e4d8412 100644 --- a/tests/test_notification_metrics.py +++ b/tests/test_notification_metrics.py @@ -1,70 +1,93 @@ from __future__ import annotations -import inspect -from typing import Any, Callable, Optional +from typing import Awaitable, Callable, Optional from unittest.mock import MagicMock import pytest import api.plugins.notifications as notifications_module from api.plugins import notifications - -WRAPPERS: list[tuple[str, str, Optional[dict[str, str]]]] = [ - ("access_request_created", "notifications.access_request_created.sent", None), - ("access_request_completed", "notifications.access_request_completed.sent", None), - ("access_expiring_user", "notifications.expiring_access.sent", {"kind": "user"}), - ("access_expiring_owner", "notifications.expiring_access.sent", {"kind": "owner"}), - ("access_expiring_role_owner", "notifications.expiring_access.sent", {"kind": "role_owner"}), - ("access_role_request_created", "notifications.role_request_created.sent", None), - ("access_role_request_completed", "notifications.role_request_completed.sent", None), - ("access_group_request_created", "notifications.group_request_created.sent", None), - ("access_group_request_completed", "notifications.group_request_completed.sent", None), +from api.plugins.notifications import NotificationHook + +# hook -> (metric name, tags) — mirrors notifications._SENT_METRICS. The +# "sent" counter is recorded by send_notification after the async hook fans out +# successfully (the async replacement for the old @hookimpl(wrapper=True) set). +HOOKS: list[tuple[NotificationHook, str, Optional[dict[str, str]]]] = [ + (NotificationHook.ACCESS_REQUEST_CREATED, "notifications.access_request_created.sent", None), + (NotificationHook.ACCESS_REQUEST_COMPLETED, "notifications.access_request_completed.sent", None), + (NotificationHook.ACCESS_EXPIRING_USER, "notifications.expiring_access.sent", {"kind": "user"}), + (NotificationHook.ACCESS_EXPIRING_OWNER, "notifications.expiring_access.sent", {"kind": "owner"}), + (NotificationHook.ACCESS_EXPIRING_ROLE_OWNER, "notifications.expiring_access.sent", {"kind": "role_owner"}), + (NotificationHook.ACCESS_ROLE_REQUEST_CREATED, "notifications.role_request_created.sent", None), + (NotificationHook.ACCESS_ROLE_REQUEST_COMPLETED, "notifications.role_request_completed.sent", None), + (NotificationHook.ACCESS_GROUP_REQUEST_CREATED, "notifications.group_request_created.sent", None), + (NotificationHook.ACCESS_GROUP_REQUEST_COMPLETED, "notifications.group_request_completed.sent", None), ] -def _kwargs_for(wrapper_fn: Callable[..., Any]) -> dict[str, Any]: - sig = inspect.signature(wrapper_fn) - return {name: True if name == "notify_requester" else None for name in sig.parameters} +class _FakeHook: + """Stand-in for pluggy's HookRelay: every hook caller returns a list holding + a single freshly-built coroutine, mimicking one registered async hookimpl.""" + + def __init__(self, behavior: Callable[[], Awaitable[None]]) -> None: + self._behavior = behavior + def __getattr__(self, name: str) -> Callable[..., list[Awaitable[None]]]: + def caller(**kwargs: object) -> list[Awaitable[None]]: + return [self._behavior()] -def _drive_success(wrapper_fn: Callable[..., Any]) -> None: - gen = wrapper_fn(**_kwargs_for(wrapper_fn)) - next(gen) - with pytest.raises(StopIteration): - gen.send(None) + return caller -def _drive_failure(wrapper_fn: Callable[..., Any], exc: BaseException) -> None: - gen = wrapper_fn(**_kwargs_for(wrapper_fn)) - next(gen) - with pytest.raises(StopIteration): - gen.throw(exc) +def _install_hook(monkeypatch: pytest.MonkeyPatch, behavior: Callable[[], Awaitable[None]]) -> None: + monkeypatch.setattr(notifications_module, "get_notification_hook", lambda: _FakeHook(behavior)) @pytest.fixture -def fake_hook(monkeypatch: pytest.MonkeyPatch) -> MagicMock: +def fake_metrics(monkeypatch: pytest.MonkeyPatch) -> MagicMock: fake = MagicMock() monkeypatch.setattr(notifications_module, "get_metrics_reporter_hook", lambda: fake) return fake -@pytest.mark.parametrize("wrapper_name,metric,tags", WRAPPERS) -def test_wrapper_emits_counter_on_success( - fake_hook: MagicMock, wrapper_name: str, metric: str, tags: Optional[dict[str, str]] +@pytest.mark.parametrize("hook,metric,tags", HOOKS) +async def test_send_notification_emits_counter_on_success( + fake_metrics: MagicMock, + monkeypatch: pytest.MonkeyPatch, + hook: NotificationHook, + metric: str, + tags: Optional[dict[str, str]], ) -> None: - _drive_success(getattr(notifications, wrapper_name)) - fake_hook.record_counter.assert_called_once_with(metric_name=metric, value=1, tags=tags) + async def ok() -> None: + return None + _install_hook(monkeypatch, ok) + await notifications.send_notification(hook) + fake_metrics.record_counter.assert_called_once_with(metric_name=metric, value=1, tags=tags) -@pytest.mark.parametrize("wrapper_name", [w[0] for w in WRAPPERS]) -def test_wrapper_does_not_emit_counter_on_failure(fake_hook: MagicMock, wrapper_name: str) -> None: - _drive_failure(getattr(notifications, wrapper_name), RuntimeError("inner plugin blew up")) - fake_hook.record_counter.assert_not_called() +@pytest.mark.parametrize("hook", [h[0] for h in HOOKS]) +async def test_send_notification_does_not_emit_on_failure( + fake_metrics: MagicMock, monkeypatch: pytest.MonkeyPatch, hook: NotificationHook +) -> None: + async def boom() -> None: + raise RuntimeError("inner plugin blew up") + + _install_hook(monkeypatch, boom) + # The plugin failure is swallowed (logged, not raised) and no counter emitted. + await notifications.send_notification(hook) + fake_metrics.record_counter.assert_not_called() -def test_metric_emit_failure_does_not_break_wrapper(monkeypatch: pytest.MonkeyPatch) -> None: - def boom() -> Any: + +async def test_metric_emit_failure_does_not_break_send(monkeypatch: pytest.MonkeyPatch) -> None: + def boom_hook() -> object: raise RuntimeError("metrics_reporter unavailable") - monkeypatch.setattr(notifications_module, "get_metrics_reporter_hook", boom) - _drive_success(notifications.access_request_created) + monkeypatch.setattr(notifications_module, "get_metrics_reporter_hook", boom_hook) + + async def ok() -> None: + return None + + _install_hook(monkeypatch, ok) + # A failure recording the metric must not propagate out of send_notification. + await notifications.send_notification(NotificationHook.ACCESS_REQUEST_CREATED) diff --git a/tests/test_role_request.py b/tests/test_role_request.py index d913e8d1..f9005db1 100644 --- a/tests/test_role_request.py +++ b/tests/test_role_request.py @@ -29,7 +29,7 @@ ModifyRoleGroups, RejectRoleRequest, ) -from api.plugins import ConditionalAccessResponse, get_conditional_access_hook, get_notification_hook +from api.plugins import ConditionalAccessResponse, get_notification_hook from api.services import okta from tests.factories import AppGroupFactory, OktaGroupFactory, OktaUserFactory, RoleGroupFactory, RoleRequestFactory from tests.helpers import db_count @@ -951,10 +951,10 @@ async def test_auto_resolve_create_role_request( notification_hook = get_notification_hook() request_created_notification_spy = mocker.patch.object(notification_hook, "access_role_request_created") request_completed_notification_spy = mocker.patch.object(notification_hook, "access_role_request_completed") - request_hook = get_conditional_access_hook() - request_created_conditional_access_spy = mocker.patch.object( - request_hook, - "role_request_created", + # Conditional-access hooks are async; patch the dispatch helper the operation + # awaits (mock.patch autospecs it to an AsyncMock since it is a coroutine fn). + request_created_conditional_access_spy = mocker.patch( + "api.operations.create_role_request.evaluate_conditional_access", return_value=[ConditionalAccessResponse(approved=True, reason="Auto-Approved")], ) add_membership_spy = mocker.patch.object(okta, "add_user_to_group") @@ -992,9 +992,8 @@ async def test_auto_resolve_create_role_request( request_created_conditional_access_spy.reset_mock() add_membership_spy.reset_mock() - request_created_conditional_access_spy = mocker.patch.object( - request_hook, - "role_request_created", + request_created_conditional_access_spy = mocker.patch( + "api.operations.create_role_request.evaluate_conditional_access", return_value=[ConditionalAccessResponse(approved=False, reason="Auto-Rejected")], ) @@ -1030,8 +1029,8 @@ async def test_auto_resolve_create_role_request( request_created_conditional_access_spy.reset_mock() add_membership_spy.reset_mock() - request_created_conditional_access_spy = mocker.patch.object( - request_hook, "role_request_created", return_value=[None] + request_created_conditional_access_spy = mocker.patch( + "api.operations.create_role_request.evaluate_conditional_access", return_value=[None] ) role_request = await CreateRoleRequest( @@ -1089,10 +1088,8 @@ async def test_auto_resolve_create_role_request_with_time_limit_constraint_tag( notification_hook = get_notification_hook() request_created_notification_spy = mocker.patch.object(notification_hook, "access_role_request_created") request_completed_notification_spy = mocker.patch.object(notification_hook, "access_role_request_completed") - request_hook = get_conditional_access_hook() - request_created_conditional_access_spy = mocker.patch.object( - request_hook, - "role_request_created", + request_created_conditional_access_spy = mocker.patch( + "api.operations.create_role_request.evaluate_conditional_access", return_value=[ ConditionalAccessResponse( approved=True, diff --git a/uv.lock b/uv.lock index 60b4084f..4f245e5f 100644 --- a/uv.lock +++ b/uv.lock @@ -72,7 +72,7 @@ requires-dist = [ { name = "itsdangerous", specifier = "==2.2.0" }, { name = "mcp", extras = ["cli"], specifier = "==1.27.0" }, { name = "okta", specifier = "==2.9.8" }, - { name = "pluggy", specifier = "==1.5.0" }, + { name = "pluggy", specifier = "==1.6.0" }, { name = "pydantic", specifier = ">=2.7,<3" }, { name = "pydantic-settings", specifier = ">=2.6,<3" }, { name = "pyjwt", extras = ["crypto"], specifier = "==2.13.0" }, @@ -1185,11 +1185,11 @@ wheels = [ [[package]] name = "pluggy" -version = "1.5.0" +version = "1.6.0" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/96/2d/02d4312c973c6050a18b314a5ad0b3210edb65a906f868e31c111dede4a6/pluggy-1.5.0.tar.gz", hash = "sha256:2cffa88e94fdc978c4c574f15f9e59b7f4201d439195c3715ca9e2486f1d0cf1", size = 67955, upload-time = "2024-04-20T21:34:42.531Z" } +sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412, upload-time = "2025-05-15T12:30:07.975Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/88/5f/e351af9a41f866ac3f1fac4ca0613908d9a41741cfcf2228f4ad853b697d/pluggy-1.5.0-py3-none-any.whl", hash = "sha256:44e1ad92c8ca002de6377e165f3e0f1be63266ab4d554740532335b9d75ea669", size = 20556, upload-time = "2024-04-20T21:34:40.434Z" }, + { url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" }, ] [[package]]