openedx_authz.engine package

openedx_authz.engine package#

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.enforcer.libraries_v2_enabled() → bool

Dummy toggle that is always enabled.

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.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#