openedx_authz.engine package

openedx_authz.engine package#

Subpackages#

Submodules#

openedx_authz.engine.adapter module#

Extended Casbin Adapter with Filtering Support.

This module provides an enhanced adapter implementation for Casbin that extends the base Django adapter with filtering capabilities. The ExtendedAdapter allows for efficient loading of policy rules from the database with support for filtering based on policy attributes.

The adapter combines functionality from both the base Adapter (for Django ORM integration) and FilteredAdapter (for selective policy loading) to provide optimized policy management for authorization systems.

class openedx_authz.engine.adapter.ExtendedAdapter(db_alias='default')

Bases: Adapter, FilteredFileAdapter

Extended Casbin adapter with filtering capabilities.

This adapter extends the base Django ORM Casbin adapter to support filtered policy loading, allowing for more efficient policy management by loading only relevant policy rules based on specified filter criteria.

Inherits from:

Adapter: Base Django adapter for Casbin policy persistence. FilteredAdapter: Interface for filtered policy loading.

filter_query(queryset: QuerySet, filter: Filter) → QuerySet

Apply filter criteria to the policy queryset.

This method takes a Django queryset of CasbinRule objects and applies filtering based on the provided filter object’s attributes. It supports filtering by policy type (ptype) and policy values (v0-v5).

Parameters:

  • queryset (QuerySet) – Django queryset of CasbinRule objects to filter.

  • filter (Filter) – Filter object with attributes (ptype, v0, v1, v2, v3, v4, v5) containing lists of values to filter by. Empty lists are ignored.

Returns:

Filtered and ordered queryset of CasbinRule objects.

Return type:

QuerySet

is_filtered() → bool

Check if the adapter supports filtering.

Returns:

True if the adapter supports filtered policy loading, False otherwise.

Return type:

bool

load_filtered_policy(model: Model, filter: Filter) → None

Load policy rules from storage with filtering applied.

This method loads policy rules from the database and applies the specified filter to load only relevant rules. The filtered rules are then loaded into the provided Casbin model.

IMPORTANT: This method is used internally by the enforcer.load_filtered_policy()

method. Do not call this method directly. If you need to load policy rules, use the enforcer.load_filtered_policy() method.

Parameters:

  • model (Model) – The Casbin model to load policy rules into.

  • filter (Filter) – Filter object containing criteria for policy selection. Should have attributes like ptype, v0, v1, etc. with lists of values to filter by.

query_policy(filter: Filter) → QuerySet

Retrieve policy rules from the database based on filter criteria.

This method constructs a Django queryset to fetch CasbinRule objects that match the specified filter attributes. It supports filtering by policy type (ptype) and policy values (v0-v5).

Parameters:

filter (Filter) – Filter object with attributes (ptype, v0, v1, v2, v3, v4, v5) containing lists of values to filter by. Empty lists are ignored.

Returns:

Queryset of CasbinRule objects matching the filter criteria.

Return type:

QuerySet

class openedx_authz.engine.adapter.PolicyAttribute(*values)

Bases: Enum

Enumeration of Casbin policy attributes.

These attributes map to the columns of the CasbinRule table, but their meaning depends on the policy type (ptype). Check the openedx_authz.engine.Filter class for more details.

PTYPE = 'ptype'

Type of policy

Type:

ptype (str)

V0 = 'v0'

First policy value.

Type:

v0 (str)

V1 = 'v1'

Second policy value.

Type:

v1 (str)

V2 = 'v2'

Third policy value.

Type:

v2 (str)

V3 = 'v3'

Fourth policy value.

Type:

v3 (str)

V4 = 'v4'

Fifth policy value.

Type:

v4 (str)

V5 = 'v5'

Sixth policy value.

Type:

v5 (str)

openedx_authz.engine.apps module#

Initialization for the casbin_adapter Django application.

This overrides the default AppConfig to avoid making queries to the database when the app is not fully loaded (e.g., while pulling translations). Moved the initialization of the enforcer to a lazy load when it’s first used.

See openedx_authz/engine/enforcer.py for the enforcer implementation.

class openedx_authz.engine.apps.CasbinAdapterConfig(app_name, app_module)

Bases: AppConfig

name = 'casbin_adapter'
ready()

Initialize the casbin_adapter app.

The upstream casbin_adapter app tries to initialize the enforcer when the app is loaded, which can lead to issues if the database is not ready (e.g., while pulling translations). To avoid this, we override the ready method and do not initialize the enforcer here.

openedx_authz.engine.enforcer module#

Core authorization enforcer for Open edX AuthZ system.

Provides a Casbin SyncedEnforcer instance with extended adapter for database policy storage and automatic policy synchronization.

Components:

  • Enforcer: Main SyncedEnforcer instance for policy evaluation

  • Adapter: ExtendedAdapter for filtered database policy loading

Usage:

from openedx_authz.engine.enforcer import AuthzEnforcer allowed = enforcer.enforce(user, resource, action)

Requires CASBIN_MODEL setting.

class openedx_authz.engine.enforcer.AuthzEnforcer

Bases: object

Singleton class to manage the Casbin SyncedEnforcer instance.

Ensures a single enforcer instance is created safely and configured with the ExtendedAdapter for policy management and automatic synchronization.

There are two main use cases for this class:

  1. Directly get the enforcer instance and initialize it if needed:

    from openedx_authz.engine.enforcer import AuthzEnforcer
enforcer = AuthzEnforcer.get_enforcer()
allowed = enforcer.enforce(user, resource, action)

  2. Instantiate the class to get the singleton enforcer instance:

    from openedx_authz.engine.enforcer import AuthzEnforcer
enforcer = AuthzEnforcer()
allowed = enforcer.get_enforcer().enforce(user, resource, action)

Any of the two approaches will yield the same singleton enforcer instance.

_enforcer

The singleton enforcer instance.

Type:

SyncedEnforcer

_adapter

The singleton adapter instance.

Type:

ExtendedAdapter

classmethod configure_enforcer_auto_loading(auto_load_policy_interval: int = None)

Enable auto-load policy and auto-save on the enforcer.

This method ensures that the singleton enforcer instance is created and ready for use.

Returns:

None

classmethod configure_enforcer_auto_save(auto_save_policy: bool)

Configure auto-save on the enforcer.

This method ensures that auto-save is enabled or disabled based on the auto_save_policy parameter.

Parameters:

auto_save_policy – True to enable auto-save, False to disable

Returns:

None

classmethod configure_enforcer_auto_save_and_load()

Enable auto-load policy and auto-save on the enforcer.

This method ensures that the singleton enforcer instance is configured for auto-load and auto-save based on settings.

Returns:

None

classmethod deactivate_enforcer()

Deactivate the current enforcer instance, if any.

This method stops the auto-load policy thread. It can be used in testing or when re-initialization of the enforcer is needed. IT DOES NOT clear the singleton instance to avoid initializing it again unintentionally.

Returns:

None

classmethod get_adapter() → ExtendedAdapter

Get the adapter instance, getting it from the enforcer if needed.

Returns:

The singleton adapter instance.

Return type:

ExtendedAdapter

classmethod get_enforcer() → SyncedEnforcer

Get the enforcer instance, creating it if needed.

Returns:

The singleton enforcer instance.

Return type:

SyncedEnforcer

classmethod invalidate_policy_cache()

Invalidate the current policy cache to force a reload on next check.

This method updates the last modified version in the cache invalidation model to a new UUID, indicating that the policy has changed.

Returns:

None

classmethod is_auto_save_enabled() → bool

Check if auto-save is currently enabled on the enforcer.

Returns:

True if auto-save is enabled, False otherwise

Return type:

bool

classmethod load_policy_if_needed()

Load policy if the last load version indicates it’s needed.

This method checks if the policy needs to be reloaded comparing the last load version with the version in the cache invalidation model, and reloads it if necessary.

Returns:

None

openedx_authz.engine.filter module#

Filter Implementation for Casbin Policy Selection.

This module provides a Filter class used to specify criteria for selective loading of Casbin policy rules. The Filter class allows for efficient policy management by enabling the loading of only relevant policy rules based on policy type and attribute values.

The Filter class is designed to work with the ExtendedAdapter to provide optimized policy loading in scenarios where only a subset of policies is needed, such as loading policies for a specific user, course, or role.

class openedx_authz.engine.filter.Filter(ptype: list[str] | None = NOTHING, v0: list[str] | None = NOTHING, v1: list[str] | None = NOTHING, v2: list[str] | None = NOTHING, v3: list[str] | None = NOTHING, v4: list[str] | None = NOTHING, v5: list[str] | None = NOTHING)

Bases: object

Filter class for selective Casbin policy loading.

This class defines filtering criteria used to load only specific policy rules from the database instead of loading all policies. Each attribute corresponds to a column in the Casbin policy storage schema and accepts a list of values to filter by.

Note

  • Empty lists for any attribute means no filtering on that attribute

  • Non-empty lists create an “IN” filter for that attribute

  • All non-empty filters are combined with AND logic

ptype: list[str] | None

Policy type filter.

  • p → Policy rule (permissions).

  • g → Grouping rule (user ↔ role).

  • g2 → Action grouping (parent action ↔ child action).

Type:

ptype (Optional[list[str]])

v0: list[str] | None

First policy value filter.

  • For p → Subject (e.g., role^org_admin, user^alice).

  • For g → User (e.g., user^alice).

  • For g2 → Parent action (e.g., act^manage).

Type:

v0 (Optional[list[str]])

v1: list[str] | None

Second policy value filter.

  • For p → Action (e.g., act^manage, act^edit).

  • For g → Role (e.g., role^org_admin).

  • For g2 → Child action (e.g., act^edit).

Type:

v1 (Optional[list[str]])

v2: list[str] | None

Third policy value filter.

  • For p → Object or resource (e.g., lib^*, org^MIT).

  • For g → Scope or resource (e.g., org^MIT).

  • For g2 → Not used.

Type:

v2 (Optional[list[str]])

v3: list[str] | None

Fourth policy value filter.

  • For p → Effect (allow or deny).

  • Otherwise unused.

Type:

v3 (Optional[list[str]])

v4: list[str] | None

Fifth policy value filter (optional additional context).

Type:

v4 (Optional[list[str]])

v5: list[str] | None

Sixth policy value filter (optional additional context).

Type:

v5 (Optional[list[str]])

openedx_authz.engine.matcher module#

Custom condition checker. Note only used for data_library scope

openedx_authz.engine.matcher.is_admin_or_superuser_check(request_user: str, request_action: str, request_scope: str) → bool

Evaluates custom, non-role-based conditions for authorization checks.

Checks attribute-based conditions that don’t rely on role assignments. Currently handles ContentLibraryData and CourseOverviewData scopes by granting access to staff and superusers.

Parameters:

  • request_user (str) – Namespaced user key (format: “user::<username>”)

  • request_action (str) – Namespaced action key (format: “action::<action_name>”)

  • request_scope (str) – Namespaced scope key (format: “scope_type::<scope_id>”)

Returns:

True if the condition is satisfied (user is staff/superuser for

ContentLibraryData and CourseOverviewData scopes), False otherwise (including when user doesn’t exist or scope type is not supported)

Return type:

bool

openedx_authz.engine.policy module#

Domain-typed policy rows and the Casbin-facing policy store (ADR 0018).

This module is the boundary between the schema-apply pipeline and Casbin. It holds two kinds of thing:

  • PolicyRow and GroupingRow — frozen, hashable value objects that model a single Casbin p (definition) row and g (assignment) row respectively. They carry the internal Casbin namespacing (role^, act^, <scope>^*) and own the translation to and from the raw list[str] form Casbin stores, delegating the field layout to the shared PolicyIndex / GroupingPolicyIndex.

  • PolicyStore — a thin wrapper over a Casbin enforcer that exposes only the policy operations the schema pipeline needs, in PolicyRow / GroupingRow terms. Callers never see the raw enforcer handle or the list[str] row shape.

Why a wrapper rather than methods on the enforcer: the schema applier runs these operations inside its own transaction.atomic() block and injects a fake in tests, so the store must (a) own enforcer resolution so callers hold no handle, (b) never open a transaction of its own, and (c) be trivially substitutable. A small wrapper satisfies all three while keeping AuthzEnforcer focused on lifecycle. The enforcer is resolved lazily on first use (or injected), so a bare PolicyStore() is safe to construct before Django settings are configured. add_*/remove_* here mutate the underlying enforcer’s in-memory model as well as the database (via its adapter); the transactional and cache-invalidation semantics remain the caller’s responsibility.

class openedx_authz.engine.policy.GroupingRow(subject: str, role: str, scope: str)

Bases: object

A single Casbin g (grouping) row: one subject-to-role assignment.

Fields follow the g shape: [subject, role, scope]. The scope segment is optional on a stored row — a scope-less assignment round-trips with an empty-string scope rather than being rejected — mirroring Casbin’s own tolerance (see parse()).

Modelling the grouping row as a typed value object lets the schema pipeline reason about assignments by name (row.role, row.subject) instead of positional indices into a raw list[str].

as_policy() → list[str]

Return the enforcer arg form: [subject, role, scope].

classmethod from_policy(values: list[str]) → GroupingRow

Build from a stored g row ([subject, role, scope]).

Parsing is delegated to the shared, strict parse(), so this stays in step with PolicyRow.from_policy(). A scope-less row ([subject, role]) is padded first, so it maps to an empty-string scope — the exact tolerance the previous positional access (row[2] if len(row) >= 3 else "") encoded inline — instead of being rejected.

role: str
scope: str
subject: str
class openedx_authz.engine.policy.PolicyRow(subject: str, action: str, scope: str, effect: str)

Bases: object

A single Casbin p row rendered from a role-permission pair.

Fields follow the p shape: subject (role), action (permission), scope pattern, effect. Namespacing to the internal Casbin form (role^, act^, <scope>^*) happens here, at the boundary — schema objects never carry those prefixes.

The Casbin policy type is a fixed "p" for every rendered definition row.

PTYPE: ClassVar[str] = 'p'
action: str
as_policy() → list[str]

Return the enforcer arg form: [subject, action, scope, effect].

effect: str
classmethod from_grant(role_id: str, permission_id: str, scope: str) → PolicyRow

Build the Casbin p row for one (role, permission, scope) grant.

The single place the internal namespacing is applied, so every producer and consumer of a rendered row agrees on its exact shape. A drift between two such places would silently stop a later comparison against the stored policy from matching anything.

classmethod from_policy(values: list[str]) → PolicyRow

Build from a stored p row ([subject, action, scope, effect]).

Parsing is delegated to the shared, strict parse(), so this stays in step with the api layer. A partially populated row is padded first, so an in-memory row round-trips instead of being rejected.

scope: str
subject: str
class openedx_authz.engine.policy.PolicyStore(enforcer=None)

Bases: object

Casbin-facing facade exposing only the operations the schema pipeline needs.

Speaks PolicyRow / GroupingRow so callers never touch the raw list[str] row shape or the enforcer handle. The underlying enforcer is resolved lazily (AuthzEnforcer.get_enforcer()) on first use unless one is injected, so a caller can build a bare PolicyStore() and let it manage the handle. Deliberately not transaction-aware: the mutating methods delegate straight to the enforcer, which mutates its in-memory model and database together, so the caller remains responsible for wrapping a batch in transaction.atomic() and for invalidating the policy cache on failure.

add_policy_row(row: PolicyRow) → bool

Add one p row. Returns the enforcer’s add result (True if added).

get_grouping_rows() → list[GroupingRow]

Return every stored g (assignment) row as a GroupingRow.

get_policy_rows() → list[PolicyRow]

Return every stored p row as a PolicyRow.

reload() → None

Reload the enforcer’s in-memory model from the database (ADR 0018 §8).

The schema applier calls this right after taking the serialization lock so the subsequent get_policy_rows read reflects the committed p rows the lock just made visible, rather than a prior run’s version-gated in-memory snapshot. Delegating through the store (instead of reaching for the global enforcer) keeps the applier’s “never hold the enforcer handle” contract and lets injected fakes model the reload.

remove_grouping_row(row: GroupingRow) → bool

Remove one g row. Returns the enforcer’s remove result (True if removed).

The row is sent back in its exact stored shape via GroupingRow.as_policy(), so the exact grouping row is removed rather than a reconstructed one.

remove_policy_row(row: PolicyRow) → bool

Remove one p row. Returns the enforcer’s remove result (True if removed).

openedx_authz.engine.renderer module#

Render compiled definitions to Casbin rows and apply them (ADR 0018 §1, §5, §8).

This is the only Casbin/Django-aware part of the schema pipeline. It implements the render and apply lifecycle steps:

  • render builds the Casbin p rows for a CompiledSchema in memory, without touching the database.

  • apply persists the rows in a single transaction, while preserving data owned by other services (ADR 0018 §3): dynamic roles, user assignments, and the legacy g2 action-inheritance rows that still live in authz.policy.

Key semantics:

  • Idempotent (ADR 0018 §2): re-applying identical definitions changes nothing and creates no duplicates. After a successful apply the stored policy equals the compiled definition — no stale rows remain.

  • Change report before write (ADR 0018 §6): SchemaApplier.plan() reports the p rows that will be added or removed by comparing rendered output against the currently stored policy.

  • Adoption, not duplication (ADR 0025 §6): a rendered row that already exists as a p row gains definition and source records without being rewritten, while a stored row no schema declares is left in place and enforceable but unattributed.

  • Removal is force-gated (ADR 0018 §6): a role slated for removal that still has user assignments requires an explicit force option. Without force the apply aborts before any write; with force the role’s p rows and its g assignment rows are removed together.

SchemaApplier.apply() reconciles the stored policy to the rendered set: it adds missing rows, removes stale rows it owns, and prunes definition/source records that the compiled schema no longer contains, all in one transaction.

The definition/source model (ADR 0018 §3, ADR 0025) is the ownership record that makes precise pruning safe, and it is consulted rather than assumed: removal candidates are intersected with the stored role-permission grants (see SchemaApplier._managed_rows()), so unmanaged p rows, dynamic roles, user assignments, and legacy g2 action inheritance are all preserved.

render is pure in behavior — it touches no database and holds no Casbin state. plan/apply resolve the enforcer only when called (not at import), so enforcer initialization still happens after Django settings are configured.

class openedx_authz.engine.renderer.ApplyResult(added: int = 0, removed: int = 0, unchanged: bool = False, plan: ChangePlan | None = None)

Bases: object

Outcome of an apply operation, for reporting.

added: int = 0
plan: ChangePlan | None = None
removed: int = 0
unchanged: bool = False
class openedx_authz.engine.renderer.ChangePlan(added_rows: list[PolicyRow] = <factory>, removed_rows: list[PolicyRow] = <factory>, unchanged: bool = False, blocking_assignments: list[tuple[str, str]]=<factory>, categories: DefinitionDiff = <factory>, permissions: DefinitionDiff = <factory>, roles: DefinitionDiff = <factory>, grants: DefinitionDiff = <factory>)

Bases: object

Diff between rendered definitions and what is currently stored.

Presented to the operator before any write (ADR 0018 §6). Covers both the Casbin p rows and the definition tables, because the apply step syncs definitions even when no policy row changes.

added_rows: list[PolicyRow]
blocking_assignments: list[tuple[str, str]]
categories: DefinitionDiff
property definition_diffs: list[tuple[str, DefinitionDiff]]

The definition diffs paired with their display label.

property definitions_unchanged: bool

True when no definition of any kind would change.

grants: DefinitionDiff
permissions: DefinitionDiff
removed_rows: list[PolicyRow]
roles: DefinitionDiff
unchanged: bool = False
class openedx_authz.engine.renderer.DefinitionDiff(added: list[str] = <factory>, updated: list[str] = <factory>, removed: list[str] = <factory>)

Bases: object

What would change for one kind of definition (ADR 0018 §6).

updated covers metadata-only edits — a new display name or icon — which change no policy row at all and would otherwise be invisible in the report.

added: list[str]
property is_empty: bool

True when this kind of definition is untouched.

removed: list[str]
updated: list[str]
class openedx_authz.engine.renderer.PolicyRenderer

Bases: object

Turns a CompiledSchema into Casbin p rows in memory.

render(schema: CompiledSchema) → RenderedPolicy

Produce one p row per (role, permission, supported scope).

Emits definition (p) rows only — never g (assignments) or g2 (action inheritance). Applies the internal Casbin namespacing here. Performs no database access. Output order is deterministic.

class openedx_authz.engine.renderer.RenderedPolicy(rows: list[PolicyRow] = <factory>)

Bases: object

The full set of p rows for a compiled schema (no DB access).

rows: list[PolicyRow]
class openedx_authz.engine.renderer.SchemaApplier(policy_store: PolicyStore | None = None)

Bases: object

Compares, then transactionally applies rendered policy to the database.

apply(rendered: RenderedPolicy, schema: CompiledSchema, *, force: bool = False) → ApplyResult

Reconcile the stored policy to the rendered set in one transaction.

Adds rendered rows not already present, removes stale schema-owned rows no longer rendered, prunes definition/source records the compiled schema no longer contains, and invalidates the policy cache so the enforcer reloads. Preserves dynamic roles, user assignments, and g2 rows.

A role slated for removal that still has user assignments is blocking: without force the apply aborts before any write; with force the role’s stale p rows and its g assignment rows are removed together (ADR 0018 §6).

If the write fails, the transaction rolls back and the policy cache is invalidated so the enforcer reloads the last committed state rather than keeping the uncommitted in-memory rows (ADR 0018 §5).

Concurrency (ADR 0018 §8): the whole read-modify-write cycle runs inside _serialized_section(), which takes an exclusive row lock as the transaction’s first statement, reloads the enforcer, and — on a change — bumps the policy cache version inside the same transaction. Two concurrent load_authz_schema runs would otherwise each diff against the same starting snapshot and both write, leaving the Casbin p rows and the Authz* definition tables disagreeing (a lost update); the lock serializes them so the second reads the already-applied state. Bumping the version in-transaction also closes the reader stale window: without it the policy writes commit before the version changes, so a concurrent enforcer could serve stale in-memory policy against already-updated rows.

Raises:

SchemaApplyError – If the plan has blocking assignments and force is False.

plan(rendered: RenderedPolicy, schema: CompiledSchema | None = None) → ChangePlan

Compute the change report without writing (ADR 0018 §6).

Compares rendered against the currently stored p rows. Flags roles that would be removed (their subject no longer appears in the rendered set) that still have user assignments as blocking.

When schema is given, the report also covers the definition tables. Apply syncs definitions even when no policy row changes, so a metadata-only edit is a real change the operator has to see — without it the report would say “unchanged” and then rewrite display metadata.

openedx_authz.engine.utils module#

Policy loader module.

This module provides functionality to load and manage policy definitions for the Open edX AuthZ system using Casbin.

class openedx_authz.engine.utils.MigrationErrorReason(*values)

Bases: StrEnum

String constants for categorising why a single role assignment failed during migration.

ASSIGNMENT_FAILED = 'assignment_failed'
NO_LEGACY_EQUIVALENT = 'no_legacy_equivalent'
NO_SCOPE = 'no_scope'
SKIPPED_FOR_FLAG_OVERRIDE = 'skipped_for_flag_override'
UNEXPECTED_ERROR = 'unexpected_error'
UNEXPECTED_SCOPE_TYPE = 'unexpected_scope_type'
UNKNOWN_ROLE = 'unknown_role'
class openedx_authz.engine.utils.MigrationMetadata(subject: str, role: str, scope: str = '', reason: MigrationErrorReason | None = None, details: str = '')

Bases: object

Normalised representation of a single role-assignment outcome during migration.

Can represent both successful and failed assignments. Populate reason / details only for failures; leave them empty for successes.

subject

External key of the user whose assignment was attempted.

Type:

str

role

Role external key (new-style for rollback, legacy key for forward).

Type:

str

scope

Scope external key, or empty string when not yet determined.

Type:

str

reason

One of the MigrationErrorReason constants; empty for successes.

Type:

openedx_authz.engine.utils.MigrationErrorReason | None

details

Optional human-readable extra context (e.g. exception message).

Type:

str

details: str = ''
reason: MigrationErrorReason | None = None
role: str
scope: str = ''
subject: str
to_dict() → dict

Convert the migration metadata to a dictionary.

openedx_authz.engine.utils.migrate_authz_to_legacy_course_roles(course_access_role_model, user_subject_model, course_id_list, org_id, delete_after_migration, excluded_course_ids: frozenset[str] = frozenset({})) → tuple[list[MigrationMetadata], list[MigrationMetadata]]

Migrate permissions from the new Casbin-based authorization model back to the legacy CourseAccessRole model. This function reads permissions from the Casbin enforcer and creates equivalent entries in the CourseAccessRole model.

This is essentially the reverse of migrate_legacy_course_roles_to_authz and is intended for rollback purposes in case of migration issues.

To build each CourseAccessRole entry, the function needs: - A user: resolved from role assignments in scopes linked to courses. - A scope: a CourseOverviewData or OrgCourseOverviewGlobData instance, optionally filtered by course_id or org_id. - A role: a role external key that maps to a legacy role in COURSE_ROLE_EQUIVALENCES.

param course_access_role_model: It should be the CourseAccessRole model. This is passed in because the function is intended to run within a Django migration context, where direct model imports can cause issues. param user_subject_model: It should be the UserSubject model. This is passed in because the function is intended to run within a Django migration context, where direct model imports can cause issues. param course_id_list: Optional list of course IDs to filter the migration. param org_id: Optional organization ID to filter the migration. param delete_after_migration: Whether to unassign successfully migrated permissions from the new model after migration. param excluded_course_ids: Course ids for which rollback is skipped (course-level override opposes the org-level transition).

openedx_authz.engine.utils.migrate_legacy_course_roles_to_authz(course_access_role_model, course_id_list, org_id, delete_after_migration, excluded_course_ids: frozenset[str] = frozenset({})) → tuple[list[MigrationMetadata], list[MigrationMetadata]]

Migrate legacy course role data to the new Casbin-based authorization model. This function reads legacy permissions from the CourseAccessRole model and assigns equivalent roles in the new authorization system.

The old Course permissions are stored in the CourseAccessRole model, it consists of the following columns:

  • user: FK to User

  • org: optional Organization string

  • course_id: optional CourseKeyField of Course

  • role: ‘instructor’ | ‘staff’ | ‘limited_staff’ | ‘data_researcher’

In the new Authz model, this would roughly translate to:

  • course_id: scope

  • user: subject

  • role: role

The scope assigned per row depends on which fields are set: - course_id set: course-level scope (e.g. “course-v1:OpenedX+CS101+2024”). - course_id blank, org set: org-level glob scope (e.g. “course-v1:OpenedX+*”). - both set: course_id takes precedence as the more specific scope.

param course_access_role_model: It should be the CourseAccessRole model. This is passed in because the function is intended to run within a Django migration context, where direct model imports can cause issues. param course_id_list: Optional list of course IDs to filter the migration. param org_id: Optional organization ID to filter the migration. param delete_after_migration: Whether to delete successfully migrated legacy permissions after migration. param excluded_course_ids: Course ids for which migration is skipped (course-level override opposes the org-level transition).

openedx_authz.engine.utils.migrate_legacy_permissions(ContentLibraryPermission)

Migrate legacy permission data to the new Casbin-based authorization model. This function reads legacy permissions from the ContentLibraryPermission model and assigns equivalent roles in the new authorization system.

The old Library permissions are stored in the ContentLibraryPermission model, it consists of the following columns:

  • library: FK to ContentLibrary

  • user: optional FK to User

  • group: optional FK to Group

  • access_level: ‘admin’ | ‘author’ | ‘read’

In the new Authz model, this would roughly translate to:

  • library: scope

  • user: subject

  • access_level: role

Now, we don’t have an equivalent concept to “Group”, for this we will go through the users in the group and assign roles independently.

param ContentLibraryPermission: The ContentLibraryPermission model to use.

openedx_authz.engine.utils.migrate_policy_between_enforcers(source_enforcer: Enforcer, target_enforcer: Enforcer) → None

Load policies from a Casbin policy file into the Django database model.

Parameters:

  • source_enforcer (Enforcer) – The Casbin enforcer instance to migrate policies from (e.g., file-based).

  • target_enforcer (Enforcer) – The Casbin enforcer instance to migrate policies to (e.g.,database).

openedx_authz.engine.utils.run_course_authoring_migration(migration_type: MigrationType, scope_type: ScopeType, scope_key: str, course_access_role_model, user_subject_model, course_id_list: list[str] | None, org_id: str | None, excluded_course_ids: frozenset[str], delete_after_migration: bool) → AuthzCourseAuthoringMigrationRun

Orchestrate a course authoring role migration with concurrency protection and lifecycle tracking.

Wraps either migrate_legacy_course_roles_to_authz() (FORWARD) or migrate_authz_to_legacy_course_roles() (ROLLBACK) with three guarantees:

  1. Concurrency guard: an AuthzCourseAuthoringMigrationRun record is created atomically before work begins. If an active run already exists for the same (scope_type, scope_key), the call is skipped immediately to prevent duplicate parallel runs.

  2. Lifecycle tracking: the run record is updated to COMPLETED, PARTIAL_SUCCESS, or FAILED regardless of the outcome, with per-role error details persisted in the record’s metadata field.

  3. Transactional safety: data-migration work runs inside an inner atomic() block so that an unexpected exception rolls back all data changes while the tracking record (updated outside that block) is always persisted.

Parameters:

  • migration_type (MigrationType) – Direction of the migration (FORWARD or ROLLBACK).

  • scope_type (ScopeType) – Granularity key dimension for the tracking record (e.g. course or org).

  • scope_key (str) – Concrete identifier — a course-v1 key or an org name.

  • course_access_role_model – The CourseAccessRole model class.

  • user_subject_model – The UserSubject model class; required for ROLLBACK, ignored otherwise.

  • course_id_list (list[str] | None) – Restrict migration to these course-v1 keys.

  • org_id (str | None) – Restrict migration to this org; takes precedence over course_id_list.

  • excluded_course_ids (frozenset[str]) – For org-scoped runs, course ids whose course-level waffle override opposes the org transition

  • delete_after_migration (bool) – Remove successfully migrated entries from the source system.

Module contents#