feat: add authz schema compilation #476
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,339 @@ | ||
| """Resolve documents into one set of static definitions (the ``compile`` step). | ||
|
|
||
| Compilation (ADR 0018 §1) merges base definitions across all documents and | ||
| applies ``role_extensions`` per ADR 0023: | ||
|
|
||
| * Extensions resolve only after every role and permission is loaded. | ||
| * An extension changes only the fields it includes; absent fields keep | ||
| their current value; it cannot change a role ID. | ||
| * Different fields from different contributions combine. | ||
| * ``priority`` resolves conflicts on the same metadata field or the same | ||
| permission (higher wins). Equal priority with disagreeing values raises | ||
| :class:`SchemaCompileError` so deployment stops before the database | ||
| changes. | ||
| * Adding a permission the role already has, or removing one it lacks, is a | ||
| no-op logged as a warning. | ||
|
|
||
| Every resulting :class:`CompiledDefinition` retains all contributing | ||
| :class:`SourceRecord` values, and each role-permission grant is attributed at | ||
| the (role, permission) grain with its origin (base vs extension) for ADR 0025 | ||
| source tracking. Output is deterministic regardless of discovery order. No | ||
| Casbin/Django imports. | ||
| """ | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| import logging | ||
| from dataclasses import dataclass, field, replace | ||
|
|
||
| from openedx_authz.constants import SchemaOriginKind | ||
| from openedx_authz.engine.schema.exceptions import SchemaCompileError | ||
| from openedx_authz.engine.schema.types import ( | ||
| CompiledDefinition, | ||
| CompiledSchema, | ||
| RelationshipSource, | ||
| RoleDefinition, | ||
| SchemaDocument, | ||
| SourceRecord, | ||
| ) | ||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
| # Metadata fields an extension may replace on a role. | ||
| _METADATA_FIELDS = ("display_name", "description", "icon", "hidden") | ||
|
|
||
| # Singular labels for operator-facing messages, keyed by document attribute. | ||
| _KIND_LABELS = {"categories": "category", "permissions": "permission", "roles": "role"} | ||
|
Comment on lines
+42
to
+46
Member
There was a problem hiding this comment. Nit: Could we apply this decorator to a shared build method, with each factory implementing only the part that differs? |
||
|
|
||
|
|
||
| @dataclass | ||
| class _Tracked: | ||
| """A base definition plus the sources and priority that produced it.""" | ||
|
|
||
| definition: object | ||
| sources: list[SourceRecord] = field(default_factory=list) | ||
| priority: int = 0 | ||
|
|
||
|
|
||
| class SchemaCompiler: | ||
| """Merges validated documents into a :class:`CompiledSchema`.""" | ||
|
|
||
| def compile(self, documents: list[SchemaDocument]) -> CompiledSchema: | ||
| """Resolve categories, permissions, roles, and extensions. | ||
|
|
||
| Assumes ``documents`` already passed validation. | ||
|
|
||
| Raises: | ||
| SchemaCompileError: On an unresolvable equal-priority conflict. | ||
| """ | ||
| categories = self._collect(documents, "categories", key=lambda c: c.id) | ||
| permissions = self._collect(documents, "permissions", key=lambda p: p.identifier) | ||
| roles = self._collect(documents, "roles", key=lambda r: r.id) | ||
|
|
||
| role_permission_sources = self._resolve_roles_and_provenance(roles, documents) | ||
|
|
||
| return CompiledSchema( | ||
| categories=self._finalize(categories), | ||
| permissions=self._finalize(permissions), | ||
| roles=self._finalize(roles), | ||
| role_permission_sources=role_permission_sources, | ||
| ) | ||
|
|
||
| # ---- base collection -------------------------------------------------- | ||
|
Member
There was a problem hiding this comment. nit: do we need these in-line comments? |
||
|
|
||
| def _collect(self, documents: list[SchemaDocument], attr: str, key) -> dict[str, _Tracked]: | ||
| """Gather base definitions keyed by identifier, resolving by priority. | ||
|
|
||
| Higher priority wins on conflict; equal priority with differing content | ||
| raises; identical duplicates merge their sources. | ||
| """ | ||
| tracked: dict[str, _Tracked] = {} | ||
| for document in documents: | ||
| for definition in getattr(document, attr): | ||
| identifier = key(definition) | ||
| existing = tracked.get(identifier) | ||
| if existing is None: | ||
| tracked[identifier] = _Tracked( | ||
| definition=definition, | ||
| sources=[document.source], | ||
| priority=document.priority, | ||
| ) | ||
| continue | ||
|
|
||
| kind = _KIND_LABELS.get(attr, attr) | ||
| if existing.definition == definition: | ||
| existing.sources.append(document.source) | ||
| elif document.priority > existing.priority: | ||
| # The loser is discarded; say so, otherwise the contributing | ||
| # file looks like it took effect (ADR 0017 §4). | ||
| self._warn_discarded( | ||
| kind, | ||
| identifier, | ||
| loser=existing.sources[0], | ||
| loser_priority=existing.priority, | ||
| winner=document.source, | ||
| winner_priority=document.priority, | ||
| ) | ||
| tracked[identifier] = _Tracked( | ||
| definition=definition, | ||
| sources=[document.source], | ||
| priority=document.priority, | ||
| ) | ||
| elif document.priority == existing.priority: | ||
| raise SchemaCompileError( | ||
| f"Conflicting {kind} definition for {identifier!r} at equal priority " | ||
| f"{document.priority} ({existing.sources[0].source_id} vs {document.source.source_id})." | ||
| ) | ||
| else: | ||
| self._warn_discarded( | ||
| kind, | ||
| identifier, | ||
| loser=document.source, | ||
| loser_priority=document.priority, | ||
| winner=existing.sources[0], | ||
| winner_priority=existing.priority, | ||
| ) | ||
| return tracked | ||
|
Comment on lines
+104
to
+136
Member
There was a problem hiding this comment. Can we move this to a different method with a single responsibility of managing priority between documents? Could it be somehow reused to manage conflicts in extensions? |
||
|
|
||
| @staticmethod | ||
| def _warn_discarded( | ||
| kind: str, | ||
| identifier: str, | ||
| *, | ||
| loser: SourceRecord, | ||
| loser_priority: int, | ||
| winner: SourceRecord, | ||
| winner_priority: int, | ||
| ) -> None: | ||
| """Report a contribution that lost to a higher-priority one. | ||
|
|
||
| Priority silently picking a winner is the behavior operators find hardest | ||
| to debug: the losing file is valid, was loaded, and simply has no effect. | ||
| ADR 0017 §4 requires warning about exactly this. | ||
| """ | ||
| logger.warning( | ||
| "authz schema: %s %r from %s (priority %s) has no effect; %s (priority %s) takes precedence.", | ||
| kind, | ||
| identifier, | ||
| loser.source_id, | ||
| loser_priority, | ||
| winner.source_id, | ||
| winner_priority, | ||
| ) | ||
|
|
||
| # ---- roles + provenance ---------------------------------------------- | ||
|
|
||
| def _resolve_roles_and_provenance( | ||
|
Member
There was a problem hiding this comment. Could we use more descriptive names for
Member
There was a problem hiding this comment. I think this should be applied retroactively to all PRs.
Member
There was a problem hiding this comment. Also can we move this to a different method with a single responsibility? Resolve _resolve_roles_and_provenance could be |
||
| self, roles: dict[str, _Tracked], documents: list[SchemaDocument] | ||
| ) -> dict[tuple[str, str], list[RelationshipSource]]: | ||
| """Apply extensions and build per-(role, permission) provenance. | ||
|
|
||
| Seeds base provenance from each role's own definition, then folds in | ||
| ``role_extensions`` (metadata replacement + permission add/remove), | ||
| honoring priority. Returns the relationship provenance map. | ||
| """ | ||
| metadata_changes, perm_changes = self._gather_extension_changes(roles, documents) | ||
| rp_sources: dict[tuple[str, str], list[RelationshipSource]] = {} | ||
|
|
||
| for role_id, tracked in roles.items(): | ||
| role: RoleDefinition = tracked.definition | ||
| base_sources = list(tracked.sources) | ||
| base_priority = tracked.priority | ||
|
|
||
| # Seed base provenance for every permission the role declares. | ||
| provenance: dict[str, list[RelationshipSource]] = { | ||
| perm: [RelationshipSource(src, SchemaOriginKind.BASE, base_priority) for src in base_sources] | ||
| for perm in role.permissions | ||
| } | ||
|
|
||
| md = metadata_changes.get(role_id, {}) | ||
| if md: | ||
| new_values, contributing_sources = self._resolve_metadata(role_id, md) | ||
| tracked.definition = replace(role, **new_values) | ||
| role = tracked.definition | ||
| for src in contributing_sources: | ||
| if src not in tracked.sources: | ||
| tracked.sources.append(src) | ||
|
|
||
| pc = perm_changes.get(role_id) | ||
| if pc and (pc["add"] or pc["remove"]): | ||
| final_perms, provenance = self._resolve_permissions( | ||
| role_id, role.permissions, base_sources, base_priority, pc | ||
| ) | ||
| tracked.definition = replace(tracked.definition, permissions=final_perms) | ||
|
|
||
| for perm, sources in provenance.items(): | ||
| rp_sources[(role_id, perm)] = sources | ||
|
|
||
| return rp_sources | ||
|
|
||
| def _gather_extension_changes(self, roles: dict[str, _Tracked], documents: list[SchemaDocument]): | ||
|
Member
There was a problem hiding this comment. Should this be split into two? |
||
| """Collect per-role metadata and permission changes from all extensions. | ||
|
|
||
| Entries carry the full :class:`SourceRecord` and priority so provenance | ||
| and conflict resolution have everything they need. | ||
| """ | ||
| metadata_changes: dict[str, dict[str, list[tuple[object, int, SourceRecord]]]] = {} | ||
| perm_changes: dict[str, dict[str, list[tuple[str, int, SourceRecord]]]] = {} | ||
|
|
||
| for document in documents: | ||
| for extension in document.role_extensions: | ||
| role_id = extension.role | ||
| if role_id not in roles: | ||
| # Validation already errors on this; skip defensively. | ||
| continue | ||
| md = metadata_changes.setdefault(role_id, {}) | ||
| for field_name in _METADATA_FIELDS: | ||
| value = getattr(extension, field_name) | ||
|
Member
There was a problem hiding this comment. Should this have a default? |
||
| if value is not None: | ||
| md.setdefault(field_name, []).append((value, document.priority, document.source)) | ||
| pc = perm_changes.setdefault(role_id, {"add": [], "remove": []}) | ||
| for perm in extension.add_permissions: | ||
| pc["add"].append((perm, document.priority, document.source)) | ||
| for perm in extension.remove_permissions: | ||
| pc["remove"].append((perm, document.priority, document.source)) | ||
| return metadata_changes, perm_changes | ||
|
|
||
| def _resolve_metadata(self, role_id: str, md: dict[str, list[tuple[object, int, SourceRecord]]]): | ||
|
Member
There was a problem hiding this comment. Can we reuse the mechanism of managing priority for all cases (base priority, this, etc)? |
||
| """Pick winning metadata values by priority; error on equal-priority ties.""" | ||
| new_values: dict[str, object] = {} | ||
| contributing: set[SourceRecord] = set() | ||
| for field_name, entries in md.items(): | ||
| max_priority = max(priority for _, priority, _ in entries) | ||
| top_values = {value for value, priority, _ in entries if priority == max_priority} | ||
| if len(top_values) > 1: | ||
| raise SchemaCompileError( | ||
| f"Conflicting {field_name!r} for role {role_id!r} at equal priority " | ||
| f"{max_priority}: {sorted(map(str, top_values))}." | ||
| ) | ||
| new_values[field_name] = next(iter(top_values)) | ||
| contributing.update(src for _, priority, src in entries if priority == max_priority) | ||
| winner = next(src for _, priority, src in entries if priority == max_priority) | ||
| for _, priority, src in entries: | ||
| if priority < max_priority: | ||
| self._warn_discarded( | ||
| f"role_extension {field_name}", | ||
| role_id, | ||
| loser=src, | ||
| loser_priority=priority, | ||
| winner=winner, | ||
| winner_priority=max_priority, | ||
| ) | ||
| return new_values, contributing | ||
|
|
||
| def _resolve_permissions( | ||
| self, | ||
| role_id: str, | ||
| base: tuple[str, ...], | ||
| base_sources: list[SourceRecord], | ||
| base_priority: int, | ||
| pc: dict[str, list[tuple[str, int, SourceRecord]]], | ||
| ): | ||
| """Apply add/remove per permission, returning (final_perms, provenance). | ||
|
|
||
| Add-vs-remove conflicts resolve by priority; equal priority raises. | ||
| Provenance keeps base attribution and appends extension attribution for | ||
| added permissions. | ||
| """ | ||
| current = set(base) | ||
| provenance: dict[str, list[RelationshipSource]] = { | ||
| perm: [RelationshipSource(src, SchemaOriginKind.BASE, base_priority) for src in base_sources] | ||
| for perm in base | ||
| } | ||
|
Member
There was a problem hiding this comment. Should we have a property for this? We also use it in https://github.com/openedx/openedx-authz/pull/476/changes#diff-752f33fcceee317067b5f683d36400f9983f027126a7ab19d38514b21b3f337fR184-R187 |
||
|
|
||
| actions: dict[str, list[tuple[str, int, SourceRecord]]] = {} | ||
| for perm, priority, src in pc["add"]: | ||
| actions.setdefault(perm, []).append(("add", priority, src)) | ||
| for perm, priority, src in pc["remove"]: | ||
| actions.setdefault(perm, []).append(("remove", priority, src)) | ||
|
Comment on lines
+284
to
+288
Member
There was a problem hiding this comment. I think we could have add/remove methods which could be called directly with a getattr, it could simplify this and the next lines |
||
|
|
||
| for perm, entries in actions.items(): | ||
| max_priority = max(priority for _, priority, _ in entries) | ||
| top = {action for action, priority, _ in entries if priority == max_priority} | ||
| if len(top) > 1: | ||
| raise SchemaCompileError( | ||
| f"Conflicting add/remove for permission {perm!r} on role {role_id!r} " | ||
| f"at equal priority {max_priority}." | ||
| ) | ||
| action = next(iter(top)) | ||
| winning_sources = [src for act, priority, src in entries if priority == max_priority and act == action] | ||
|
|
||
| for act, priority, src in entries: | ||
| if priority < max_priority: | ||
| self._warn_discarded( | ||
| f"role_extension {act} of {perm!r} on role", | ||
| role_id, | ||
| loser=src, | ||
| loser_priority=priority, | ||
| winner=winning_sources[0], | ||
| winner_priority=max_priority, | ||
| ) | ||
|
Comment on lines
+299
to
+310
Member
There was a problem hiding this comment. Same question about reusability |
||
|
|
||
| if action == "add": | ||
| if perm in current: | ||
| logger.warning("role_extension adds %r already on role %r; no-op.", perm, role_id) | ||
| current.add(perm) | ||
| provenance.setdefault(perm, []) | ||
| provenance[perm].extend( | ||
| RelationshipSource(src, SchemaOriginKind.EXTENSION, max_priority) for src in winning_sources | ||
| ) | ||
| else: # remove | ||
| if perm not in current: | ||
| logger.warning("role_extension removes %r not on role %r; no-op.", perm, role_id) | ||
| current.discard(perm) | ||
| provenance.pop(perm, None) | ||
|
|
||
| return tuple(sorted(current)), provenance | ||
|
|
||
| # ---- finalize --------------------------------------------------------- | ||
|
|
||
| def _finalize(self, tracked: dict[str, _Tracked]) -> dict[str, CompiledDefinition]: | ||
| """Turn tracked definitions into CompiledDefinition entries.""" | ||
| return { | ||
| identifier: CompiledDefinition( | ||
| key=identifier, | ||
| definition=entry.definition, | ||
| sources=tuple(entry.sources), | ||
| ) | ||
| for identifier, entry in tracked.items() | ||
| } | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can this also be a class instead of a constant so we keep following the same patterns as before?