Add global default authorizers for core workflows

Author: Ben Elam

orchestrator/settings.py

@@ -17,10 +17,12 @@
 from typing import Literal
 
 from pydantic import Field, NonNegativeInt, PostgresDsn, RedisDsn
+from pydantic.main import BaseModel
 from pydantic_settings import BaseSettings
 
 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 +113,28 @@ 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
+
+
+_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.internal_authorize_callback,
+    retry_auth_callback=authorizers.internal_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.internal_authorize_callback,
+    retry_auth_callback=authorizers.internal_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.internal_authorize_callback,
+    retry_auth_callback=authorizers.internal_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.internal_authorize_callback,
+    retry_auth_callback=authorizers.internal_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.internal_authorize_callback,
+    retry_auth_callback=authorizers.internal_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.internal_authorize_callback,
+    retry_auth_callback=authorizers.internal_retry_auth_callback,
+)
 def task_validate_products() -> StepList:
     return (
         init