Add global default authorizers for core-internal workflows

docs/reference-docs/auth-backend-and-frontend.md

@@ -34,7 +34,7 @@ WFO provides authentication based on an OIDC provider. The OIDC provider is pres
 
 #### Frontend
 
-The WFO frontend uses [NextAuth](3) to handle authentication. Authentication configuration can be found in [page/api/auth/[...nextauth].ts](4)
+The WFO frontend uses [NextAuth][3] to handle authentication. Authentication configuration can be found in [page/api/auth/[...nextauth].ts][4]
 
 **ENV variables**
 These variables need to be set for authentication to work on the frontend.
@@ -56,7 +56,7 @@ NEXTAUTH_SECRET=[SECRET] // Used by NextAuth to encrypt the JWT token
 
 With authentication turned on and these variables provided the frontend application will redirect unauthorized users to the login screen provided by the OIDC provider to request their credentials and return them to the page they tried to visit.
 
-Note: It's possible to add additional oidc providers including some that are provided by the NextAuth library like Google, Apple and others. See [NextAuthProviders](5) for more information.
+Note: It's possible to add additional oidc providers including some that are provided by the NextAuth library like Google, Apple and others. See [NextAuthProviders][5] for more information.
 
 ##### Authorization
 
@@ -219,7 +219,9 @@ class GraphqlAuthorization(ABC):
 
 ```
 
-Graphql Authorization decisions can be made based on request properties and user attributes
+Graphql Authorization decisions can be made based on request properties and user attributes.
+
+[Additional methods](#authorization-for-internal-workflows) exist for defining role-based access control for internal workflows.
 
 ### Example
 
@@ -525,6 +527,24 @@ are prioritized in different workflow and inputstep configurations.
   </tbody>
 </table>
 
+### Authorization for internal workflows
+Users of Workflow Orchestrator can't directly access the `@workflow` decorators of tasks and workflows defined within `orchestrator-core`.
+However, authorization callbacks can still be passed via the `OrchestratorCore` class when initializing your WFO application.
+
+```python
+from orchestrator import OrchestratorCore
+
+app = OrchestratorCore()
+
+# Let foo and bar be Authorizers
+app.register_internal_authorize_callback(foo)
+app.register_internal_retry_auth_callback(bar)
+```
+
+If these callbacks are not registered, these workflows can be started and retried by all users by default.
+
+For more on application startup, see the [Settings Overview page][settings-overview].
+
 ### Examples
 Assume we have the following function that can be used to create callbacks:
 
@@ -590,6 +610,8 @@ We can now construct a variety of authorization policies.
 
     Note that we could specify `auth=allow_roles("user")` if helpful, or we can omit `auth` to fail open to any logged in user.
 
+[settings-overview]: ../../reference-docs/app/settings-overview
+
 [1]: https://github.com/workfloworchestrator/example-orchestrator-ui
 [2]: https://github.com/workfloworchestrator/example-orchestrator
 [3]: https://next-auth.js.org/

orchestrator/app.py

@@ -57,7 +57,8 @@
 from orchestrator.log_config import LOGGER_OVERRIDES
 from orchestrator.metrics import ORCHESTRATOR_METRICS_REGISTRY, initialize_default_metrics
 from orchestrator.services.process_broadcast_thread import ProcessDataBroadcastThread
-from orchestrator.settings import AppSettings, ExecutorType, app_settings
+from orchestrator.settings import AppSettings, ExecutorType, app_settings, get_authorizers
+from orchestrator.utils.auth import Authorizer
 from orchestrator.version import GIT_COMMIT_HASH
 from orchestrator.websocket import init_websocket_manager
 from pydantic_forms.exception_handlers.fastapi import form_error_handler
@@ -311,6 +312,38 @@ def register_graphql_authorization(self, graphql_authorization_instance: Graphql
         """
         self.auth_manager.graphql_authorization = graphql_authorization_instance
 
+    def register_internal_authorize_callback(self, callback: Authorizer) -> None:
+        """Registers the authorize_callback for WFO's internal workflows and tasks.
+
+        Since RBAC policies are applied to workflows via decorator, this enables registration of callbacks
+        for workflows defined in orchestrator-core itself.
+        However, this assignment MUST be made before any workflows are run.
+
+        Args:
+            callback (Authorizer): The async Authorizer to run for the `authorize_callback` argument of internal workflows.
+
+        Returns:
+            None
+        """
+        authorizers = get_authorizers()
+        authorizers.internal_authorize_callback = callback
+
+    def register_internal_retry_auth_callback(self, callback: Authorizer) -> None:
+        """Registers the retry_auth_callback for WFO's internal workflows and tasks.
+
+        Since RBAC policies are applied to workflows via decorator, this enables registration of callbacks
+        for workflows defined in orchestrator-core itself.
+        However, this assignment MUST be made before any workflows are run.
+
+        Args:
+            callback (Authorizer): The async Authorizer to run for the `retry_auth_callback` argument of internal workflows.
+
+        Returns:
+            None
+        """
+        authorizers = get_authorizers()
+        authorizers.internal_retry_auth_callback = callback
+
 
 main_typer_app = typer.Typer()
 main_typer_app.add_typer(cli_app, name="orchestrator", help="The orchestrator CLI commands")

orchestrator/graphql/schemas/process.py

@@ -89,8 +89,8 @@ async def user_permissions(self, info: OrchestratorInfo) -> FormUserPermissionsT
         auth_resume, auth_retry = get_auth_callbacks(get_steps_to_evaluate_for_rbac(process), workflow)
 
         return FormUserPermissionsType(
-            retryAllowed=bool(auth_retry and auth_retry(oidc_user)),
-            resumeAllowed=bool(auth_resume and auth_resume(oidc_user)),
+            retryAllowed=bool(auth_retry and await auth_retry(oidc_user)),
+            resumeAllowed=bool(auth_resume and await auth_resume(oidc_user)),
         )
 
     @authenticated_field(description="Returns list of subscriptions of the process")  # type: ignore

orchestrator/settings.py

@@ -17,10 +17,13 @@
 from typing import Literal
 
 from pydantic import Field, NonNegativeInt, PostgresDsn, RedisDsn
+from pydantic.main import BaseModel
 from pydantic_settings import BaseSettings
 
+from oauth2_lib.fastapi import OIDCUserModel
 from oauth2_lib.settings import oauth2lib_settings
 from orchestrator.services.settings_env_variables import expose_settings
+from orchestrator.utils.auth import Authorizer
 from orchestrator.utils.expose_settings import SecretStr as OrchSecretStr
 from pydantic_forms.types import strEnum
 
@@ -111,3 +114,48 @@ class AppSettings(BaseSettings):
     expose_settings("app_settings", app_settings)  # type: ignore
 if app_settings.EXPOSE_OAUTH_SETTINGS:
     expose_settings("oauth2lib_settings", oauth2lib_settings)  # type: ignore
+
+
+class Authorizers(BaseModel):
+    # Callbacks specifically for orchestrator-core callbacks.
+    # Separate from defaults for user-defined workflows and steps.
+    internal_authorize_callback: Authorizer | None = None
+    internal_retry_auth_callback: Authorizer | None = None
+
+    async def authorize_callback(self, user: OIDCUserModel | None) -> bool:
+        """This is the authorize_callback to be registered for workflows defined within orchestrator-core.
+
+        If Authorizers.internal_authorize_callback is None, this function will return True.
+        i.e. any user will be authorized to start internal workflows.
+        """
+        if self.internal_authorize_callback is None:
+            return True
+        return await self.internal_authorize_callback(user)
+
+    async def retry_auth_callback(self, user: OIDCUserModel | None) -> bool:
+        """This is the retry_auth_callback to be registered for workflows defined within orchestrator-core.
+
+        If Authorizers.internal_retry_auth_callback is None, this function will return True.
+        i.e. any user will be authorized to retry internal workflows on failure.
+        """
+        if self.internal_retry_auth_callback is None:
+            return True
+        return await self.internal_retry_auth_callback(user)
+
+
+_authorizers = Authorizers()
+
+
+def get_authorizers() -> Authorizers:
+    """Acquire singleton of app authorizers to assign these callbacks at app setup.
+
+    Ensures downstream users can acquire singleton without being tempted to do
+    from orchestrator.settings import authorizers
+    authorizers = my_authorizers
+    or
+    from orchestrator import settings
+    settings.authorizers = my_authorizers
+
+    ...each of which goes wrong in its own way.
+    """
+    return _authorizers

orchestrator/workflows/modify_note.py

@@ -13,6 +13,7 @@
 from orchestrator.db import db
 from orchestrator.forms import SubmitFormPage
 from orchestrator.services import subscriptions
+from orchestrator.settings import get_authorizers
 from orchestrator.targets import Target
 from orchestrator.utils.json import to_serializable
 from orchestrator.workflow import StepList, done, init, step, workflow
@@ -21,6 +22,8 @@
 from pydantic_forms.types import FormGenerator, State, UUIDstr
 from pydantic_forms.validators import LongText
 
+authorizers = get_authorizers()
+
 
 def initial_input_form(subscription_id: UUIDstr) -> FormGenerator:
     subscription = subscriptions.get_subscription(subscription_id)
@@ -51,6 +54,12 @@ def store_subscription_note(subscription_id: UUIDstr, note: str) -> State:
     }
 
 
-@workflow("Modify Note", initial_input_form=wrap_modify_initial_input_form(initial_input_form), target=Target.MODIFY)
+@workflow(
+    "Modify Note",
+    initial_input_form=wrap_modify_initial_input_form(initial_input_form),
+    target=Target.MODIFY,
+    authorize_callback=authorizers.authorize_callback,
+    retry_auth_callback=authorizers.retry_auth_callback,
+)
 def modify_note() -> StepList:
     return init >> store_process_subscription() >> store_subscription_note >> done

orchestrator/workflows/removed_workflow.py

@@ -12,11 +12,18 @@
 # limitations under the License.
 
 
+from orchestrator.settings import get_authorizers
 from orchestrator.workflow import StepList, workflow
 
+authorizers = get_authorizers()
+
 
 # This workflow has been made to create the initial import process for a SN7 subscription
 # it does not do anything but is needed for the correct showing in the GUI.
-@workflow("Dummy workflow to replace removed workflows")
+@workflow(
+    "Dummy workflow to replace removed workflows",
+    authorize_callback=authorizers.authorize_callback,
+    retry_auth_callback=authorizers.retry_auth_callback,
+)
 def removed_workflow() -> StepList:
     return StepList()

orchestrator/workflows/tasks/cleanup_tasks_log.py

@@ -17,12 +17,14 @@
 from sqlalchemy import select
 
 from orchestrator.db import ProcessTable, db
-from orchestrator.settings import app_settings
+from orchestrator.settings import app_settings, get_authorizers
 from orchestrator.targets import Target
 from orchestrator.utils.datetime import nowtz
 from orchestrator.workflow import ProcessStatus, StepList, done, init, step, workflow
 from pydantic_forms.types import State
 
+authorizers = get_authorizers()
+
 
 @step("Clean up completed tasks older than TASK_LOG_RETENTION_DAYS")
 def remove_tasks() -> State:
@@ -41,6 +43,11 @@ def remove_tasks() -> State:
     return {"tasks_removed": count}
 
 
-@workflow("Clean up old tasks", target=Target.SYSTEM)
+@workflow(
+    "Clean up old tasks",
+    target=Target.SYSTEM,
+    authorize_callback=authorizers.authorize_callback,
+    retry_auth_callback=authorizers.retry_auth_callback,
+)
 def task_clean_up_tasks() -> StepList:
     return init >> remove_tasks >> done

orchestrator/workflows/tasks/resume_workflows.py

@@ -17,10 +17,12 @@
 
 from orchestrator.db import ProcessTable, db
 from orchestrator.services import processes
+from orchestrator.settings import get_authorizers
 from orchestrator.targets import Target
 from orchestrator.workflow import ProcessStatus, StepList, done, init, step, workflow
 from pydantic_forms.types import State, UUIDstr
 
+authorizers = get_authorizers()
 logger = structlog.get_logger(__name__)
 
 
@@ -110,6 +112,8 @@ def restart_created_workflows(created_state_process_ids: list[UUIDstr]) -> State
 @workflow(
     "Resume all workflows that are stuck on tasks with the status 'waiting', 'created' or 'resumed'",
     target=Target.SYSTEM,
+    authorize_callback=authorizers.authorize_callback,
+    retry_auth_callback=authorizers.retry_auth_callback,
 )
 def task_resume_workflows() -> StepList:
     return init >> find_waiting_workflows >> resume_found_workflows >> restart_created_workflows >> done

orchestrator/workflows/tasks/validate_product_type.py

@@ -25,10 +25,12 @@
     get_validation_product_workflows_for_subscription,
     start_validation_workflow_for_workflows,
 )
+from orchestrator.settings import get_authorizers
 from orchestrator.targets import Target
 from orchestrator.workflow import StepList, done, init, step, workflow
 from pydantic_forms.types import FormGenerator, State
 
+authorizers = get_authorizers()
 logger = structlog.get_logger(__name__)
 
 
@@ -86,7 +88,11 @@ def validate_product_type(product_type: str) -> State:
 
 
 @workflow(
-    "Validate all subscriptions of Product Type", target=Target.SYSTEM, initial_input_form=initial_input_form_generator
+    "Validate all subscriptions of Product Type",
+    target=Target.SYSTEM,
+    initial_input_form=initial_input_form_generator,
+    authorize_callback=authorizers.authorize_callback,
+    retry_auth_callback=authorizers.retry_auth_callback,
 )
 def task_validate_product_type() -> StepList:
     return init >> validate_product_type >> done

orchestrator/workflows/tasks/validate_products.py

@@ -26,12 +26,15 @@
 from orchestrator.services.products import get_products
 from orchestrator.services.translations import generate_translations
 from orchestrator.services.workflows import get_workflow_by_name, get_workflows
+from orchestrator.settings import get_authorizers
 from orchestrator.targets import Target
 from orchestrator.utils.errors import ProcessFailureError
 from orchestrator.utils.fixed_inputs import fixed_input_configuration as fi_configuration
 from orchestrator.workflow import StepList, done, init, step, workflow
 from pydantic_forms.types import State
 
+authorizers = get_authorizers()
+
 # Since these errors are probably programming failures we should not throw AssertionErrors
 
 
@@ -187,7 +190,12 @@ def check_subscription_models() -> State:
     return {"check_subscription_models": True}
 
 
-@workflow("Validate products", target=Target.SYSTEM)
+@workflow(
+    "Validate products",
+    target=Target.SYSTEM,
+    authorize_callback=authorizers.authorize_callback,
+    retry_auth_callback=authorizers.retry_auth_callback,
+)
 def task_validate_products() -> StepList:
     return (
         init

orchestrator/workflows/tasks/validate_subscriptions.py

@@ -24,7 +24,7 @@
     get_validation_product_workflows_for_subscription,
     start_validation_workflow_for_workflows,
 )
-from orchestrator.settings import app_settings
+from orchestrator.settings import app_settings, get_authorizers
 from orchestrator.targets import Target
 from orchestrator.workflow import StepList, init, step, workflow
 
@@ -33,6 +33,8 @@
 
 task_semaphore = BoundedSemaphore(value=2)
 
+authorizers = get_authorizers()
+
 
 @step("Validate subscriptions")
 def validate_subscriptions() -> None:
@@ -56,6 +58,11 @@ def validate_subscriptions() -> None:
         start_validation_workflow_for_workflows(subscription=subscription, workflows=validation_product_workflows)
 
 
-@workflow("Validate subscriptions", target=Target.SYSTEM)
+@workflow(
+    "Validate subscriptions",
+    target=Target.SYSTEM,
+    authorize_callback=authorizers.authorize_callback,
+    retry_auth_callback=authorizers.retry_auth_callback,
+)
 def task_validate_subscriptions() -> StepList:
     return init >> validate_subscriptions