Query Permission Patterns¶
Reference for how OpenContracts filters querysets by user permissions.
Architecture Overview¶
Permission-filtered access is organised in three tiers:
Tier 0 Model managers user_can() / visible_to_user()
(authorization — single-object + queryset gates
primitives) — internal foundation, NOT a public entry
Tier 1 Service packages opencontractserver/<app>/services/
(THE public layer) — get_* / list_* / create_* / update_* / delete_*
— every user-context caller goes here
— BaseService.{get_or_none, filter_visible,
filter_visible_qs,
require_permission, user_has}
provides the generic surface
Tier 1.5 Optimizers Private prefetch / bulk-permission helpers
(internal detail) — live INSIDE services; not referenced
outside their owning service
Rule of thumb after Phase 6 (issue #1720): resolvers, MCP/LLM tools, REST views, and user-context Celery tasks never touch Tier 0 directly. They call Tier 1. The invariant is enforced for config/graphql/ by opencontractserver/tests/architecture/test_graphql_service_layer.py and should be extended to other consumer directories as they migrate.
Migration Recipes — "I just got an opencontracts.E001 error, what do I type?"¶
If you landed here from a Django manage.py error or a pytest failure mentioning opencontracts.E001 / "Tier-0 permission primitive", the recipe below is the answer. Both the test and the system check route through the same helper (opencontractserver.shared.architecture_audit. format_violation), so the failure output you saw is the same content as what follows — copy whichever recipe matches the identifier the scanner flagged.
In every recipe: add the import once at the top of the file:
from opencontractserver.shared.services.base import BaseService
And always pass request=info.context (or the request-equivalent your caller has) so the Tier-2 permission cache is shared across the request. (Caveat: the cache is keyed by single object, so request= only does anything on the single-object helpers — get_or_none / require_permission / user_has. On the queryset helpers filter_visible / filter_visible_qs it is accepted for API parity but is currently a no-op; pass it anyway for consistency.)
Recipe 1 — visible_to_user¶
# ❌ Forbidden
Model.objects.visible_to_user(user)
Model.objects.visible_to_user(user).get(pk=id)
# ✅ Listing visible rows (queryset, chainable like before)
BaseService.filter_visible(Model, user, request=info.context)
# ✅ IDOR-safe single-object fetch — returns None instead of raising;
# collapses not-found and permission-denied into one branch
obj = BaseService.get_or_none(Model, pk, user, request=info.context)
if obj is None:
return MyMutation(ok=False, message="Not found")
The "raises DoesNotExist on miss" form is intentionally gone. Returning None is the IDOR-safe contract — callers must surface a single generic error string for both "missing" and "forbidden" cases so the two branches stay indistinguishable to the client. If the surrounding code relied on DoesNotExist, replace the try/except with a if obj is None: branch returning the same error.
Recipe 2 — user_can¶
# ❌ Forbidden
if not obj.user_can(user, PermissionTypes.UPDATE):
return MyMutation(ok=False, message="Denied")
# ✅ Fail-fast gate — returns "" on grant, error string on denial
error = BaseService.require_permission(
obj, user, PermissionTypes.UPDATE, request=info.context
)
if error:
return MyMutation(ok=False, message=error)
# ❌ Forbidden (boolean UI flag)
can_edit = obj.user_can(user, PermissionTypes.UPDATE)
# ✅ Boolean for UI-state fields (can_edit_summary, can_view_history, etc.)
can_edit = BaseService.user_has(
obj, user, PermissionTypes.UPDATE, request=info.context
)
require_permission is for resolvers that need to abort with an error message. user_has is for resolvers that need a True/False to feed a UI-state GraphQL field. Don't reach for not BaseService.require_permission(...) when you can use BaseService.user_has(...) directly.
Recipe 3 — user_has_permission_for_obj¶
Same as user_can. Use BaseService.require_permission(...) for a gate or BaseService.user_has(...) for a boolean.
When the generic helpers aren't enough¶
If a dedicated per-app service method exists that matches your operation semantically (e.g. CorpusDocumentService.get_corpus_document_by_id, ConversationService.get_threads_for_corpus, AnnotationService.get_document_annotations), prefer it — it may layer prefetches, request-scoped caching, or extra domain logic on top of the generic BaseService.* call. The per-app service inventory below lists every one.
If your operation is a complex multi-object flow that doesn't fit either bucket, that's a signal to add a new method to the relevant per-app service rather than re-composing inline. The point of the rule is that permission-laden orchestration stops being "the same six lines copy-pasted everywhere" and starts being named, testable, importable operations.
The BaseService entry point¶
opencontractserver.shared.services.base.BaseService is the generic surface every per-app service inherits. It centralises the four cross-cutting operations that resolvers / tools used to inline against Tier 0:
| Method | Use for | Returns |
|---|---|---|
BaseService.filter_visible(Model, user, *, request=None, **kwargs) | Listing a model's visible rows. **kwargs are passed straight through to the manager's visible_to_user(...) so per-model perf knobs (e.g. Document(..., lightweight=True)) keep working. | QuerySet |
BaseService.filter_visible_qs(queryset, user, *, request=None, **kwargs) | Same intent as filter_visible, but starts from an existing queryset or related manager — chains the queryset's own visible_to_user so the visibility predicate stays in the same WHERE expression tree as any prior .filter(...) clauses (no correlated pk__in subquery). Use this in DjangoObjectType.get_queryset overrides and reverse-relation field resolvers. | QuerySet |
BaseService.get_or_none(Model, pk, user, permission=None, *, request=None) | IDOR-safe single-object fetch — returns None for both not-found and not-permitted so callers cannot enumerate via differential errors. | Instance | None |
BaseService.require_permission(instance, user, permission, *, request=None, error_message=None) | Fail-fast gate on a single object. Returns "" on grant; a human-readable denial string otherwise — feed it straight into a ServiceResult / ok=False envelope. ⚠️ Truthy on denial (inverse of the legacy if user_can(...) idiom — see the in-method docstring). | str |
BaseService.user_has(instance, user, permission, *, request=None) | Boolean yes/no for UI-state fields (can_edit_summary, can_create_labels, etc.). | bool |
Resolvers should always pass request=info.context so the Tier-2 permission cache (PermissionQueryOptimizer) is shared across the request. The cache is keyed by single object, so request= is only effective on the single-object helpers (get_or_none / require_permission / user_has); on filter_visible / filter_visible_qs it is accepted for API parity but is currently a no-op.
Tier-0 primitives (internal foundation)¶
visible_to_user(user) and user_can(user, instance, permission) live on managers/querysets. They are the source of truth for "can this user see / do this?", but they are not the recommended entry point for application code. They power BaseService.* and per-app service methods.
| Model | Implementation | File |
|---|---|---|
| Corpus | PermissionedTreeQuerySet.visible_to_user | opencontractserver/shared/QuerySets.py |
| Document | DocumentQuerySet.visible_to_user / BaseVisibilityManager.visible_to_user | opencontractserver/shared/QuerySets.py, opencontractserver/shared/Managers.py |
| Annotation / Note / Relationship | *QuerySet.visible_to_user | opencontractserver/shared/QuerySets.py |
| UserFeedback | UserFeedbackQuerySet.visible_to_user | opencontractserver/shared/QuerySets.py |
| Fallback | PermissionQuerySet.visible_to_user | opencontractserver/shared/QuerySets.py |
The single-object check is Model.objects.user_can(user, obj, permission) / obj.user_can(user, permission) — paired with visible_to_user and pinned to agree by opencontractserver/tests/permissioning/test_authorization_invariants.py.
Per-app service inventory (Tier 1)¶
| App | Service module | Key methods |
|---|---|---|
corpuses | services/corpus_service.py (CorpusService) | delete_corpus, set_visibility, update_description, grant_creator_permissions |
corpuses | services/corpus_documents.py (CorpusDocumentService) | get_corpus_documents (corpus-as-gate), get_corpus_documents_visible_to_user (MIN-perm), get_corpus_document_by_id/by_slug, is_document_in_corpus, doc add/remove |
corpuses | services/corpus_groups.py (CorpusGroupService) | list_visible_groups, get_group_by_id/get_group_by_ref (IDOR-safe, pk or slug), get_group_corpora_visible_to_user (call-time member-corpus gate for search_across_corpora), group CRUD |
corpuses | services/folders.py (FolderCRUDService) | get_visible_folders, get_folder_by_id, get_folder_tree, folder CRUD |
corpuses | services/folder_documents.py (FolderDocumentService) | get_folder_documents, move_document_to_folder, get_document_folder |
corpuses | services/lifecycle.py (DocumentLifecycleService) | get_deleted_documents, soft_delete_document, restore_document, permanently_delete_document, empty_trash |
corpuses | services/paths.py (CorpusPathService) | DocumentPath disambiguation internals |
documents | document_service.py (DocumentService) | get_document_by_id, create_document, set_document_permissions, check_user_upload_quota, validate_file_type |
documents | services/actions.py (DocumentActionsService) | get_document_actions, get_extracts_for_document, get_analysis_rows_for_document |
documents | services/relationships.py (DocumentRelationshipService) | get_visible_relationships, get_relationships_for_document, get_relationship_counts_by_document, user_has_permission |
documents | services/versions.py (DocumentVersionService) | get_version_counts_by_tree |
annotations | services/annotation_service.py (AnnotationService) | get_document_annotations, get_corpus_annotations, get_extract_annotation_summary |
annotations | services/relationship_service.py (RelationshipService) | get_document_relationships, get_relationship_summary |
analyzer | services/analysis_service.py (AnalysisService) | check_analysis_permission, get_visible_analyses, get_analysis_annotations |
analyzer | services/analysis_lifecycle_service.py (AnalysisLifecycleService) | make_public, start_document_analysis, delete_analysis |
extracts | services/extract_service.py (ExtractService) | check_extract_permission, get_visible_extracts, get_extract_datacells |
extracts | services/metadata.py (MetadataService) | get_corpus_metadata_columns, get_document_metadata, get_documents_metadata_batch, check_metadata_mutation_permission |
conversations | services/conversation_service.py (ConversationService) | get_threads_for_corpus, get_threads_for_document, get_chats_for_user, check_conversation_visibility, get_corpus_conversation_counts |
users | services/user_service.py (UserService) | get_visible_users, check_user_visibility, get_users_for_mention |
badges | services/badge_service.py (BadgeService) | get_visible_user_badges, check_user_badge_visibility, get_badges_for_user |
agents | services/agent_configuration_service.py (AgentConfigurationService) | list_visible_agents, search_mentionable_agents, get_active_agents_by_slugs, get_agent_by_id, agent CRUD |
agents | services/agent_action_result_service.py (AgentActionResultService) | list_visible_results |
notifications | services/notification_service.py (NotificationService) | list_for_user, get_for_user, unread_count, mark_read / mark_unread / mark_all_read, delete_for_user |
feedback | services/user_feedback_service.py (UserFeedbackService) | approve_annotation, reject_annotation |
worker_uploads | services/corpus_access_token_service.py (CorpusAccessTokenService) | list_for_corpus, create_token, revoke_token |
worker_uploads | services/worker_account_service.py (WorkerAccountService) | list_visible_accounts, create_worker_account, set_active |
worker_uploads | services/worker_document_upload_service.py (WorkerDocumentUploadService) | list_for_corpus |
Permission models by object type¶
| Object | Own Permissions | Inherited From | Pattern |
|---|---|---|---|
| Corpus | Guardian (direct) | — | read_corpus via corpususerobjectpermission |
| CorpusGroup | Guardian (direct) | — | read_corpusgroup via corpusgroupuserobjectpermission; member corpora additionally gated per-user at read time (CorpusGroupService.get_group_corpora_visible_to_user) |
| Document | Guardian (direct) | — | read_document via documentuserobjectpermission |
| Annotation | None (inherited) | Document + Corpus | MIN(document_permission, corpus_permission) |
| Relationship | None (inherited) | Document + Corpus | Same as Annotation |
| Note | None (inherited) | Document + Corpus | Same as Annotation |
| Analysis | Hybrid | Own + Corpus | Own guardian permissions + corpus visibility |
| Extract | Hybrid | Own + Corpus | Own guardian permissions + corpus visibility |
| Conversation | Simple | Corpus + Document | Corpus and document visibility checks |
Key patterns¶
MIN(document, corpus) Permission Inheritance¶
Annotations, relationships, and notes have no individual guardian permissions. Visibility is determined by the intersection of document and corpus visibility. Implementation: opencontractserver/shared/QuerySets.py (AnnotationQuerySet, NoteQuerySet).
Guardian Permission Table Lookup¶
For models with direct guardian permissions, query the {model}userobjectpermission table directly instead of using get_objects_for_user for better performance. Implementation: opencontractserver/shared/Managers.py, opencontractserver/shared/QuerySets.py.
Request-Level Caching¶
ConversationService and other Tier-1 services cache corpus and document visibility subqueries per request to avoid repeated permission checks.
Two-Tier user_can Caching (issue #1640)¶
Authorization checks via the centralized Manager.user_can / obj.user_can / _default_user_can API are cached at two layers so repeated checks within a single request collapse to one DB hit:
- Tier 1 — Per-instance memoization.
get_users_permissions_for_objcaches its result on the instance as_oc_granted_perms_cache: dict, keyed by(user_id, include_group_permissions_bool) → frozenset[str]. Transparent to every caller ofuser_can. - Tier 2 — Request-scoped optimizer.
PermissionQueryOptimizerlives on the GraphQL request asrequest._permission_query_optimizer, keyed by(user_id, content_type_id, instance_pk, include_group_permissions). Engaged by passingrequest=(anyBaseServicemethod threads it through for you).
Invalidation contract: set_permissions_for_obj_to_user(..., request=...) clears both tiers for the affected (user, instance) pair when called inside the HTTP lifecycle. Out-of-band changes (raw assign_perm/remove_perm, migrations, Celery tasks reusing instances, group-permission changes) bypass this hook — callers must invalidate manually via delattr(instance, INSTANCE_PERMS_CACHE_ATTR) and/or get_request_optimizer(request).invalidate(user_id=..., instance=...).
Implementation: opencontractserver/utils/permission_optimizer.py, opencontractserver/utils/permissioning.py:get_users_permissions_for_obj, opencontractserver/utils/permissioning.py:_default_user_can.
Tests: opencontractserver/tests/permissioning/test_permission_optimizer.py.
Analysis/Extract Privacy Filtering¶
Annotations created by analyses or extracts inherit visibility from those parent objects. If a user cannot see the analysis/extract, they cannot see its annotations. Implementation: opencontractserver/shared/QuerySets.py (AnnotationQuerySet).
IDOR Protection¶
Mutations use BaseService.get_or_none with unified error messages to prevent object ID enumeration. The helper returns None for both not-found and not-permitted, so the caller surfaces a single string that does not leak which condition failed.
Structural Annotation Handling¶
Structural annotations (headers, sections) are always visible if the parent document is visible, regardless of creator. They are read-only for non-superusers. Implementation: opencontractserver/shared/QuerySets.py (AnnotationQuerySet visibility_filter).
Enforcement¶
See docs/development/architecture_invariants.md for the invariant index and the pattern for adding new ones.
Two independent layers point at the same scanner (opencontractserver/shared/architecture_audit.py) so a violation cannot slip through either:
- Django system check (
opencontractserver/shared/checks.py, registered viausers/apps.py:ready()). Fires on every management command —manage.py runserver,migrate,shell,test,check --deploy. Emitsopencontracts.E001and blocks the command with a non-zero exit code on any inline Tier-0 use. This is the "fail on startup" guardrail — devs see violations immediately on the firstmanage.pyinvocation, not only when CI runs pytest. - Pytest invariant (
opencontractserver/tests/architecture/test_graphql_service_layer.py). Runs in CI on every push, alongsidetest_authorization_invariantsandtest_security_hardening. Pins the same scanner plus a regression test that the Django check stays registered.
The allowlist (opencontractserver.shared.architecture_audit.ALLOWED_FILES) is empty. Every config/graphql/ module is scanned (recursively, including subpackages such as permissioning/permission_annotator/) — filters.py no longer needs an entry because its only remaining references to forbidden identifiers are inside comments (which the AST scanner already ignores).
Enforcement scope (important): both layers scan config/graphql/ only. The "always go through services" rule is policy for the other user-context surfaces — MCP tools (opencontractserver/mcp/), LLM tools (opencontractserver/llms/tools/), REST views, and user-context Celery tasks — but the scanner does not cover them yet, and they still contain correct-but-inline Tier-0 calls. A green opencontracts.E001 means "no inline Tier-0 in config/graphql/," not "none anywhere." Expanding the scan to those packages (after migrating their existing inline usage) is tracked as future work.
When you add a new resolver/mutation/MCP tool/REST view that needs permission-filtered access, either:
- Call an existing dedicated service method (preferred when one matches your operation semantically).
- Call
BaseService.{get_or_none, filter_visible, filter_visible_qs, require_permission, user_has}directly for generic "list visible X" / "chain visibility onto an existing queryset" / "fetch X for user" / "gate write on X" / "boolean has-perm-on-X" operations.
Do not import Tier-0 identifiers into consumer code.
See Also¶
docs/permissioning/consolidated_permissioning_guide.md— full permissioning architecturedocs/architecture/sharing.md— object sharing patternsdocs/refactor_plans/2026-05-19-service-layer-centralization-design.md— the design that produced this layering