Skip to content

msflib.scope

Part of the msflib core package.

msflib.scope

Unified scope system for MSFLib.

Provides a canonical, extensible scope model for multi-tenant data access control.

ScopeCompiler handles generic scope transformations (to string, segments, or mapping). Domain-specific transformations (e.g., checkpoint keys, retrieval filters, storage keys) belong in the subsystem classes that use them (e.g., CheckpointScope, factories).

Quick start::

from msflib.scope import (
    ScopeEnvelope,
    PrincipalContext,
    ScopeCompiler,
    DimensionRegistry,
    create_default_registry,
)

registry = create_default_registry()
compiler = ScopeCompiler(registry=registry)

envelope = ScopeEnvelope(
    dimensions={"tenant_id": "42", "workspace_id": "7"},
    principal=PrincipalContext(account_id="user-1"),
)

# Compile to canonical scope string
scope_string = compiler.compile_scope_string(envelope, operation="retrieval.search")

# Compile to canonical scope segments tuple
segments = compiler.compile_scope_segments(envelope, operation="retrieval.search")

# Compile to canonical scope mapping
scope_map = compiler.compile_scope_map(envelope, operation="retrieval.search")

ScopeCompiler(registry: DimensionRegistry | None = None, profile_registry: ProfileRegistry | None = None, string_delimiter: str = '/', string_style: str = 'key_value_pairs', sentinel: str = _NONE_SENTINEL, lock_registry: bool = False)

Translates a :class:ScopeEnvelope into various sink-specific formats.

Parameters:

Name Type Description Default
registry DimensionRegistry | None

Dimension registry used to resolve dimension specs.

 None
profile_registry ProfileRegistry | None

Operation profile registry used for validation.

 None

register_profile(name: str, profile: dict[str, Any]) -> None

Register (or replace) an operation profile by name.

get_profile(name: str) -> dict[str, Any]

Return the profile dict for name.

validate_profile(envelope: ScopeEnvelope, profile_name: str) -> None

Validate that envelope satisfies profile_name.

Raises :exc:~msflib.scope.errors.ProfileValidationError on failure.

get_operation_policy_config(operation: str) -> dict[str, Any]

Return policy-related profile settings for operation.

This keeps policy evaluation concerns outside sink projection.

compile_scope_map(envelope: ScopeEnvelope, *, operation: str, use_sentinel: bool = True) -> dict[str, str] | dict[str, str | None] 

compile_scope_map(envelope: ScopeEnvelope, *, operation: str, use_sentinel: Literal[True] = True) -> dict[str, str]

compile_scope_map(envelope: ScopeEnvelope, *, operation: str, use_sentinel: Literal[False]) -> dict[str, str | None]

Return canonical scope as an ordered key/value mapping.

Pass use_sentinel=False to keep null/absent required_nullable dimensions as None instead of substituting the compiler's sentinel — for consumers that need to distinguish "explicitly null" from a real value without comparing against the sentinel string.

compile_scope_segments(envelope: ScopeEnvelope, *, operation: str) -> tuple[str, ...]

Return canonical scope as flat ordered segments: (k1, v1, k2, v2, ...).

compile_scope_string(envelope: ScopeEnvelope, *, operation: str, delimiter: str | None = None, style: str | None = None, order: tuple[str, ...] | None = None) -> str

Return canonical scope as a delimited flat segment string.

compile_scope_items(envelope: ScopeEnvelope, *, operation: str, use_sentinel: bool = True) -> tuple[tuple[str, str], ...] | tuple[tuple[str, str | None], ...] 

compile_scope_items(envelope: ScopeEnvelope, *, operation: str, use_sentinel: Literal[True] = True) -> tuple[tuple[str, str], ...]

compile_scope_items(envelope: ScopeEnvelope, *, operation: str, use_sentinel: Literal[False]) -> tuple[tuple[str, str | None], ...]

Build canonical scope items from a profile-derived contract.

Contract semantics: - required: must be present and non-null. - required_nullable: always projected, with null/absent replaced by sentinel (or left as None when use_sentinel=False). - optional: projected only when the envelope actually supplies a non-null value for that dimension; absent optional dimensions are omitted entirely (no sentinel filler).

ConstraintViolationError(message: str, decision: ScopeDecision | None = None)

Bases: ScopeError

Raised when a constraint evaluator denies access.

Attach the :class:~msflib.scope.model.ScopeDecision for structured error handling without re-evaluating the constraint chain.

DimensionConflictError

Bases: DimensionError

Raised when a duplicate or conflicting dimension is registered.

DimensionError

Bases: ScopeError

Base for dimension registration and lookup errors.

ProfileValidationError

Bases: ScopeError

Raised when a scope does not satisfy an operation profile's requirements.

ScopeError

Bases: Exception

Base exception for all scope-related errors.

ScopeValidationError

Bases: ScopeError

Raised when scope dimensions fail validation.

Typically indicates missing or malformed required dimensions before compiling to a sink-specific format.

UnknownDimensionError

Bases: DimensionError

Raised when a dimension key or alias is not in the registry.

DimensionSpec(canonical_key: str, aliases: frozenset[str] = frozenset(), normalizer: Callable[[Any], str] | None = None, validator: Callable[[str | None], bool] | None = None, key_alias: str | None = None, priority: int = 0, sensitive: bool = False) dataclass

Metadata for a single scope dimension.

Attributes:

Name Type Description
canonical_key str

Primary name used in :class:ScopeEnvelope dimensions.
aliases frozenset[str]

Alternative names that resolve to this dimension.
normalizer Callable[[Any], str] | None

Callable (Any) -> str that converts non-None raw values to canonical strings. Defaults to :func:_default_normalizer. Note: the normalizer is not called for None values; None is treated as a special case and passed through by :meth:normalize.
validator Callable[[str | None], bool] | None

Optional callable (str | None) -> bool for extra validation after normalization.
key_alias str | None

Optional compact key name for string serialization.
priority int

Higher values appear first in DimensionRegistry.list_all().
sensitive bool

Sensitive dimensions are redacted in audit summaries.

normalize(value: Any) -> str | None

Return the normalised string value, or None if value is None.

validate_value(value: str | None) -> bool

Return True if value passes the spec's validator (if any).

PrincipalContext(account_id: str, roles: frozenset[str] = frozenset(), claims: Mapping[str, Any] = dict(), request_id: str | None = None, timestamp: datetime = (lambda: datetime.now(timezone.utc))()) dataclass

Immutable representation of the authenticated caller.

Attributes:

Name Type Description
account_id str

The caller's system-wide account identifier (user).
roles frozenset[str]

Frozenset of role names granted to this principal.
claims Mapping[str, Any]

Arbitrary JWT/session claims forwarded from the auth layer.
request_id str | None

Optional tracing identifier for the originating request.
timestamp datetime

When the principal context was captured.

has_role(role: str) -> bool

Return True if role is in this principal's role set.

ScopeDecision(allowed: bool, reason_code: str = '', constraint_name: str = '', metadata: Mapping[str, Any] = dict()) dataclass

Result of evaluating one or more :class:ScopeConstraint instances.

Attributes:

Name Type Description
allowed bool

True when access is permitted.
reason_code str

Machine-readable code describing the outcome.
constraint_name str

Name of the constraint that produced this decision.
metadata Mapping[str, Any]

Optional structured data about the decision.

allow(*, reason_code: str = 'allowed', constraint_name: str = '', metadata: dict[str, Any] | None = None) -> ScopeDecision classmethod

Factory for an allow decision.

deny(*, reason_code: str, constraint_name: str = '', metadata: dict[str, Any] | None = None) -> ScopeDecision classmethod

Factory for a deny decision.

to_dict() -> dict[str, Any]

Serialize decision to dict for logging/events.

ScopeEnvelope(dimensions: Mapping[str, str | None], attributes: Mapping[str, Any] = dict(), principal: PrincipalContext | None = None, metadata: Mapping[str, Any] = dict()) dataclass

Canonical, immutable scope context for a single operation.

dimensions holds normalised values keyed by canonical dimension name. A dimension key that is absent means "not specified". A dimension key present with a None value means "explicitly null" (e.g. workspace_id=None is the workspace-less bucket per decision D3).

Attributes:

Name Type Description
dimensions Mapping[str, str | None]

Mapping of canonical dimension key → normalised string (or None).
attributes Mapping[str, Any]

Non-dimension metadata (e.g. operation hints, flags).
principal PrincipalContext | None

Authenticated caller context; may be None for service calls.
metadata Mapping[str, Any]

Request-level tracing and observability metadata.

get_dim(key: str) -> str | None

Return the normalised value for key, or None if absent or null.

has_dim(key: str) -> bool

Return True if key is present (even with a None value).

to_dict(*, registry: DimensionRegistry | None = None, redact_sensitive: bool = False) -> dict[str, Any]

Serialise to a plain dict (suitable for JSON / logging).

Parameters:

Name Type Description Default
registry DimensionRegistry | None

Optional dimension registry used to determine sensitive dimensions for redaction.

 None
redact_sensitive bool

If True, replace sensitive dimension values with "<redacted>" when registry is provided.

 False

canonical_items(*, order: tuple[str, ...], sentinel: str | None) -> tuple[tuple[str, str | None], ...]

Return ordered key/value scope pairs with sentinel substitution.

This provides a canonical shape that can be transformed into any sink representation by simple joins/splits. Pass sentinel=None to keep null/absent values as None instead of substituting a filler.

canonical_segments(*, order: tuple[str, ...], sentinel: str) -> tuple[str, ...]

Return flat ordered segments: (k1, v1, k2, v2, ...).

from_dict(data: dict[str, Any], *, registry: DimensionRegistry | None = None) -> ScopeEnvelope classmethod

Construct from a plain dict, optionally resolving aliases via registry.

When registry is provided, aliases are resolved to canonical keys and values are normalised using each dimension's normalizer function. Without a registry, values are still validated via _default_normalizer (which rejects booleans, empty strings, and collections).

ChannelConstraint(required_channel: str, *, deny_without_channel: bool = True)

Validates that the request channel matches the required channel.

Channel is read from scope.attributes['channel'] first and falls back to context['current_channel'] when absent.

ConstraintEvaluator(constraints: list[ScopeConstraint] | None = None, enforcement_level: str = 'log-only')

Evaluates an ordered list of constraints and returns a composite decision.

Constraints are evaluated in order until the first deny. The enforcement_level controls what happens after that deny:

- ``log-only``: return *allow* (violations are logged but not enforced).
- ``shadow``: return *deny* (violations are logged; callers observe the
    denial without an exception being raised).
- ``enforce``: raise :class:`ConstraintViolationError` on the first deny.

Parameters:

Name Type Description Default
constraints list[ScopeConstraint] | None

Ordered list of :class:ScopeConstraint instances.

 None
enforcement_level str

One of "log-only", "shadow", "enforce". Defaults to "log-only" (D5 locked decision).

 'log-only'

last_decisions: list[ScopeDecision] property

Return decisions from the most recent :meth:evaluate call.

evaluate(scope: ScopeEnvelope, context: dict[str, Any] | None = None) -> ScopeDecision

Run all constraints in order; return first deny or final allow.

Side-effects: - Stores decisions in :attr:last_decisions for audit. - Logs violations at WARNING level. - Emits scope.decision events and records metrics for the first deny (including enforce raises) and for the final effective decision (an unconditional allow, a shadow deny, or a log-only override). A log-only override therefore emits and records both the original deny and the overridden allow.

The enforcement_level controls what happens on a deny: - log-only: Log and return allow. - shadow: Log and return deny (but do not raise). - enforce: Log and raise :class:ConstraintViolationError.

DimensionMatchConstraint(required_match: dict[str, str] | None = None, *, constraint_name: str | None = None)

Checks that dimension values in the scope match those in context.

Pass required_match as a mapping of {canonical_dim_key: context_key}. If the context key is present and the scope has that dimension, the values must match exactly. Missing context keys are skipped (not treated as deny).

Example::

constraint = DimensionMatchConstraint(
    required_match={"tenant_id": "current_tenant_id"},
)

If scope.get_dim("tenant_id") != context["current_tenant_id"], the constraint returns a deny decision.

MembershipConstraint(*, deny_without_principal: bool = True, fail_open: bool = False)

Stub: validates that the principal is a member of the scoped workspace.

Production implementations should inject a membership service or DB session via the context dict (key: membership_service).

By default (fail_open=False), missing or misconfigured services fail closed (deny) for security. Set fail_open=True to allow in shadow/log-only modes when the service is unavailable (e.g., for testing or during deployment).

If no principal is attached and the scope targets a concrete workspace, this constraint denies by default to avoid unauthenticated bypasses.

RoleConstraint(required_role: str, *, deny_without_principal: bool = True)

Stub: validates that the principal has the required role.

Requires the scope envelope to carry a PrincipalContext with a roles set. The required_role is checked against principal.roles.

If no principal is attached, behaviour depends on deny_without_principal: - True (default): deny. - False: allow (useful for service-to-service calls where role is implicit).

ScopeConstraint

Bases: Protocol

Protocol that every scope constraint must implement.

Implementations should be stateless (or thread-safe) so they can be shared across requests.

evaluate(scope: ScopeEnvelope, context: dict[str, Any]) -> ScopeDecision

Evaluate the constraint against scope and context.

Parameters:

Name Type Description Default
scope ScopeEnvelope

The immutable scope envelope for the current operation.

 required
context dict[str, Any]

Runtime lookup context (e.g. DB session, service clients, request headers). Constraints must not mutate it.

 required

Returns:

Name Type Description
A ScopeDecision

class:~msflib.scope.model.ScopeDecision indicating allow/deny.

ScopeDecisionMetrics(total_allowed: int = 0, total_denied: int = 0, by_constraint: dict[str, int] = (lambda: defaultdict(int))(), by_reason_code: dict[str, int] = (lambda: defaultdict(int))(), by_enforcement_level: dict[str, int] = (lambda: defaultdict(int))()) dataclass

Aggregated metrics for scope decision outcomes.

Attributes:

Name Type Description
total_allowed int

Count of decisions with allowed=True.
total_denied int

Count of decisions with allowed=False.
by_constraint dict[str, int]

Breakdown of denials by constraint name.
by_reason_code dict[str, int]

Breakdown of denials by reason_code.
by_enforcement_level dict[str, int]

Breakdown of denials by enforcement_level.

to_dict() -> dict[str, Any]

Serialize metrics to dict.

ScopeMetricsCollector()

Thread-safe collector for scope decision metrics.

Can be used to track decisions across all ConstraintEvaluator instances for observability and debugging.

record_decision(decision: ScopeDecision, enforcement_level: str = 'log-only') -> None

Record a scope decision outcome.

get_metrics() -> ScopeDecisionMetrics

Get a snapshot of current metrics.

reset() -> None

Reset all metrics to zero.

ProfileRegistry(*, load_built_ins: bool = True)

Manages operation profiles used for scope validation.

Usage::

registry = ProfileRegistry()
registry.register(
    "my_op",
    {
        "required": ["tenant_id"],
        "required_nullable": ["workspace_id"],
        "optional": [],
    },
)
profile = registry.get("my_op")

Built-in profiles (retrieval.search, memory.put, etc.) are automatically loaded at construction time.

register(name: str, profile: dict[str, Any]) -> None

Register (or replace) a profile.

Parameters:

Name Type Description Default
name str

Operation name (e.g. "documents.ingest").

 required
profile dict[str, Any]

Dict with required, required_nullable, optional (and optionally constraints, enforcement_level keys).

 required

Raises:

Type Description
ValueError

If name is empty or profile is not a dict.
ProfileValidationError

If profile declares overlapping required and required_nullable dimensions.

get(name: str) -> dict[str, Any]

Return the profile for name.

Raises:

Type Description
ProfileValidationError

If name is not registered.

has(name: str) -> bool

Return True if name is registered.

list_all() -> list[str]

Return sorted list of registered profile names.

DimensionRegistry()

Central registry of scope dimension specifications.

Usage::

registry = DimensionRegistry()
registry.register(DimensionSpec(canonical_key="project_id", priority=25))
registry.lock()

canonical = registry.resolve_alias("project_id")  # "project_id"
value = registry.normalize("project_id", 42)      # "42"

After :meth:lock is called, :meth:register raises :exc:RuntimeError.

register(spec: DimensionSpec) -> None

Register spec. Raises on duplicate canonical key or alias conflict.

get(key: str) -> DimensionSpec

Return the spec for key (canonical or alias).

Raises :exc:~msflib.scope.errors.UnknownDimensionError if not found.

resolve_alias(key: str) -> str

Return the canonical key for key (which may be canonical, alias, or key_alias).

Raises :exc:~msflib.scope.errors.UnknownDimensionError if unknown.

has(key: str) -> bool

Return True if key is a known canonical key, alias, or key_alias.

list_all() -> list[DimensionSpec]

Return all registered specs sorted by priority (highest first).

normalize(key: str, value: Any) -> str | None

Normalise value using the spec for key.

Returns None when value is None (explicit null).

validate(key: str, value: str | None) -> bool

Return True if value passes the spec's validator for key.

lock() -> None

Lock the registry. Further register() calls will raise.

is_locked() -> bool

Return True if the registry has been locked.

build_context_scope(*, tenant_id: int | None = None, workspace_id: int | None = None, account_id: int | None = None) -> ScopeEnvelope

Build the caller's context ScopeEnvelope from tier identifiers.

Each id comes from a get_current_*-style dependency (see get_scope_dependencies); entity-specific dimensions are layered on later by domain code.

get_scope_dependencies(*, get_current_tenant: Callable | None = None, get_current_workspace: Callable | None = None, get_current_active_user: Callable | None = None, get_current_account: Callable | None = None) -> DependencyNamespace

Return reusable FastAPI dependencies yielding the caller's ScopeEnvelope.

One factory takes the site's callables and returns a DependencyNamespace (the module deps convention, see auth.get_account_dependencies). Each member is a distinct named scope shape — pick the one matching the tiers an endpoint acts in rather than parameterizing a single dependency.

get_current_tenant resolves the caller's tenant the same way get_current_workspace/get_current_account resolve theirs: as a request-scoped FastAPI dependency returning a model instance (.id read off it below), validated at the endpoint boundary instead of a bare ambient string threaded in from deep inside action/service code. Pass msflib.tenancy.deps.get_tenant_dependencies(...).get_current_tenant once modules/tenancy is mounted. Omit it (the default) for a site with no tenant concept yet -- every scope below then carries no tenant_id dimension at all.

Which identity callables to pass reduces to one question: does this site have a membership entity distinct from the account — a per-workspace row with its own status and role, resolved via auth.get_user_dependencies?

  • Yes — pass get_current_workspace + get_current_active_user. get_current_active_user is what actually verifies membership (the account_id/workspace_id join and per-workspace active/kicked/banned status live there — see auth.get_current_user); workspace resolution alone does not check it, so this pair cannot be replaced by a bare get_current_account + get_current_workspace. get_account_scope then carries all three tiers, and get_workspace_scope is also available.

  • No — pass get_current_account (from auth.get_account_dependencies — typically get_current_active_account, so an inactive account can't reach a scoped endpoint). Whether a workspace tier exists at all is then a second, independent choice:

  • No workspace concept: omit get_current_workspace. Only get_account_scope / get_tenant_scope are exposed — get_workspace_scope has no meaning without workspaces.

  • Workspace exists as an attribute of the account (e.g. an Account.current_workspace_id field), not a separate membership row: also pass get_current_workspace. get_workspace_scope becomes available too, since a workspace tier now exists — but nothing here verifies the account belongs to that workspace, because there is no membership row to check membership against.

Returns:

Type Description
DependencyNamespace exposing:
  • get_account_scope — every tier the caller has (tenant + account, plus workspace when the site has one): for member-facing reads and writes where user/account-scoped rows participate.
  • get_workspace_scope — tenant + workspace, no account tier: for admin surfaces that only touch workspace-scoped rows (an account_id there would fetch user-scoped rows only to discard them). Only present when the site has a workspace tier.
  • get_tenant_scope — tenant only, for platform-level surfaces.

get_scope_metrics() -> ScopeDecisionMetrics

Get current scope decision metrics.

reset_scope_metrics() -> None

Reset metrics (useful for testing).

create_default_registry() -> DimensionRegistry

Create a new :class:DimensionRegistry pre-loaded with built-in dimensions.

compiler

Scope compiler: translates a :class:ScopeEnvelope to sink-specific formats.

Each compile_* method is deterministic and derived from the same canonical ordered scope shape to keep transformation costs low.

ScopeCompiler(registry: DimensionRegistry | None = None, profile_registry: ProfileRegistry | None = None, string_delimiter: str = '/', string_style: str = 'key_value_pairs', sentinel: str = _NONE_SENTINEL, lock_registry: bool = False)

Translates a :class:ScopeEnvelope into various sink-specific formats.

Parameters:

Name Type Description Default
registry DimensionRegistry | None

Dimension registry used to resolve dimension specs.

 None
profile_registry ProfileRegistry | None

Operation profile registry used for validation.

 None
register_profile(name: str, profile: dict[str, Any]) -> None

Register (or replace) an operation profile by name.

get_profile(name: str) -> dict[str, Any]

Return the profile dict for name.

validate_profile(envelope: ScopeEnvelope, profile_name: str) -> None

Validate that envelope satisfies profile_name.

Raises :exc:~msflib.scope.errors.ProfileValidationError on failure.

get_operation_policy_config(operation: str) -> dict[str, Any]

Return policy-related profile settings for operation.

This keeps policy evaluation concerns outside sink projection.

compile_scope_map(envelope: ScopeEnvelope, *, operation: str, use_sentinel: bool = True) -> dict[str, str] | dict[str, str | None] 
compile_scope_map(envelope: ScopeEnvelope, *, operation: str, use_sentinel: Literal[True] = True) -> dict[str, str]

compile_scope_map(envelope: ScopeEnvelope, *, operation: str, use_sentinel: Literal[False]) -> dict[str, str | None]

Return canonical scope as an ordered key/value mapping.

Pass use_sentinel=False to keep null/absent required_nullable dimensions as None instead of substituting the compiler's sentinel — for consumers that need to distinguish "explicitly null" from a real value without comparing against the sentinel string.

compile_scope_segments(envelope: ScopeEnvelope, *, operation: str) -> tuple[str, ...]

Return canonical scope as flat ordered segments: (k1, v1, k2, v2, ...).

compile_scope_string(envelope: ScopeEnvelope, *, operation: str, delimiter: str | None = None, style: str | None = None, order: tuple[str, ...] | None = None) -> str

Return canonical scope as a delimited flat segment string.

compile_scope_items(envelope: ScopeEnvelope, *, operation: str, use_sentinel: bool = True) -> tuple[tuple[str, str], ...] | tuple[tuple[str, str | None], ...] 
compile_scope_items(envelope: ScopeEnvelope, *, operation: str, use_sentinel: Literal[True] = True) -> tuple[tuple[str, str], ...]

compile_scope_items(envelope: ScopeEnvelope, *, operation: str, use_sentinel: Literal[False]) -> tuple[tuple[str, str | None], ...]

Build canonical scope items from a profile-derived contract.

Contract semantics: - required: must be present and non-null. - required_nullable: always projected, with null/absent replaced by sentinel (or left as None when use_sentinel=False). - optional: projected only when the envelope actually supplies a non-null value for that dimension; absent optional dimensions are omitted entirely (no sentinel filler).

deps

Request-scoped FastAPI dependencies yielding the caller's ScopeEnvelope.

Endpoints that act on an entire scope should pull it in as a function parameter via these dependencies rather than assembling envelopes in endpoint bodies — the chosen shape then shows up in the endpoint signature.

build_context_scope(*, tenant_id: int | None = None, workspace_id: int | None = None, account_id: int | None = None) -> ScopeEnvelope

Build the caller's context ScopeEnvelope from tier identifiers.

Each id comes from a get_current_*-style dependency (see get_scope_dependencies); entity-specific dimensions are layered on later by domain code.

get_scope_dependencies(*, get_current_tenant: Callable | None = None, get_current_workspace: Callable | None = None, get_current_active_user: Callable | None = None, get_current_account: Callable | None = None) -> DependencyNamespace

Return reusable FastAPI dependencies yielding the caller's ScopeEnvelope.

One factory takes the site's callables and returns a DependencyNamespace (the module deps convention, see auth.get_account_dependencies). Each member is a distinct named scope shape — pick the one matching the tiers an endpoint acts in rather than parameterizing a single dependency.

get_current_tenant resolves the caller's tenant the same way get_current_workspace/get_current_account resolve theirs: as a request-scoped FastAPI dependency returning a model instance (.id read off it below), validated at the endpoint boundary instead of a bare ambient string threaded in from deep inside action/service code. Pass msflib.tenancy.deps.get_tenant_dependencies(...).get_current_tenant once modules/tenancy is mounted. Omit it (the default) for a site with no tenant concept yet -- every scope below then carries no tenant_id dimension at all.

Which identity callables to pass reduces to one question: does this site have a membership entity distinct from the account — a per-workspace row with its own status and role, resolved via auth.get_user_dependencies?

  • Yes — pass get_current_workspace + get_current_active_user. get_current_active_user is what actually verifies membership (the account_id/workspace_id join and per-workspace active/kicked/banned status live there — see auth.get_current_user); workspace resolution alone does not check it, so this pair cannot be replaced by a bare get_current_account + get_current_workspace. get_account_scope then carries all three tiers, and get_workspace_scope is also available.

  • No — pass get_current_account (from auth.get_account_dependencies — typically get_current_active_account, so an inactive account can't reach a scoped endpoint). Whether a workspace tier exists at all is then a second, independent choice:

  • No workspace concept: omit get_current_workspace. Only get_account_scope / get_tenant_scope are exposed — get_workspace_scope has no meaning without workspaces.

  • Workspace exists as an attribute of the account (e.g. an Account.current_workspace_id field), not a separate membership row: also pass get_current_workspace. get_workspace_scope becomes available too, since a workspace tier now exists — but nothing here verifies the account belongs to that workspace, because there is no membership row to check membership against.

Returns:

Type Description
DependencyNamespace exposing:
  • get_account_scope — every tier the caller has (tenant + account, plus workspace when the site has one): for member-facing reads and writes where user/account-scoped rows participate.
  • get_workspace_scope — tenant + workspace, no account tier: for admin surfaces that only touch workspace-scoped rows (an account_id there would fetch user-scoped rows only to discard them). Only present when the site has a workspace tier.
  • get_tenant_scope — tenant only, for platform-level surfaces.

errors

Exceptions for the unified scope system.

ScopeError

Bases: Exception

Base exception for all scope-related errors.

ScopeValidationError

Bases: ScopeError

Raised when scope dimensions fail validation.

Typically indicates missing or malformed required dimensions before compiling to a sink-specific format.

ConstraintViolationError(message: str, decision: ScopeDecision | None = None)

Bases: ScopeError

Raised when a constraint evaluator denies access.

Attach the :class:~msflib.scope.model.ScopeDecision for structured error handling without re-evaluating the constraint chain.

DimensionError

Bases: ScopeError

Base for dimension registration and lookup errors.

DimensionConflictError

Bases: DimensionError

Raised when a duplicate or conflicting dimension is registered.

UnknownDimensionError

Bases: DimensionError

Raised when a dimension key or alias is not in the registry.

ProfileValidationError

Bases: ScopeError

Raised when a scope does not satisfy an operation profile's requirements.

model

Core data models for the unified scope system.

Models are frozen dataclasses. Mapping fields are defensively copied and wrapped in read-only proxies so callers cannot mutate top-level scope, principal, or decision mappings in place.

DimensionSpec(canonical_key: str, aliases: frozenset[str] = frozenset(), normalizer: Callable[[Any], str] | None = None, validator: Callable[[str | None], bool] | None = None, key_alias: str | None = None, priority: int = 0, sensitive: bool = False) dataclass

Metadata for a single scope dimension.

Attributes:

Name Type Description
canonical_key str

Primary name used in :class:ScopeEnvelope dimensions.
aliases frozenset[str]

Alternative names that resolve to this dimension.
normalizer Callable[[Any], str] | None

Callable (Any) -> str that converts non-None raw values to canonical strings. Defaults to :func:_default_normalizer. Note: the normalizer is not called for None values; None is treated as a special case and passed through by :meth:normalize.
validator Callable[[str | None], bool] | None

Optional callable (str | None) -> bool for extra validation after normalization.
key_alias str | None

Optional compact key name for string serialization.
priority int

Higher values appear first in DimensionRegistry.list_all().
sensitive bool

Sensitive dimensions are redacted in audit summaries.
normalize(value: Any) -> str | None

Return the normalised string value, or None if value is None.

validate_value(value: str | None) -> bool

Return True if value passes the spec's validator (if any).

PrincipalContext(account_id: str, roles: frozenset[str] = frozenset(), claims: Mapping[str, Any] = dict(), request_id: str | None = None, timestamp: datetime = (lambda: datetime.now(timezone.utc))()) dataclass

Immutable representation of the authenticated caller.

Attributes:

Name Type Description
account_id str

The caller's system-wide account identifier (user).
roles frozenset[str]

Frozenset of role names granted to this principal.
claims Mapping[str, Any]

Arbitrary JWT/session claims forwarded from the auth layer.
request_id str | None

Optional tracing identifier for the originating request.
timestamp datetime

When the principal context was captured.
has_role(role: str) -> bool

Return True if role is in this principal's role set.

ScopeDecision(allowed: bool, reason_code: str = '', constraint_name: str = '', metadata: Mapping[str, Any] = dict()) dataclass

Result of evaluating one or more :class:ScopeConstraint instances.

Attributes:

Name Type Description
allowed bool

True when access is permitted.
reason_code str

Machine-readable code describing the outcome.
constraint_name str

Name of the constraint that produced this decision.
metadata Mapping[str, Any]

Optional structured data about the decision.
allow(*, reason_code: str = 'allowed', constraint_name: str = '', metadata: dict[str, Any] | None = None) -> ScopeDecision classmethod

Factory for an allow decision.

deny(*, reason_code: str, constraint_name: str = '', metadata: dict[str, Any] | None = None) -> ScopeDecision classmethod

Factory for a deny decision.

to_dict() -> dict[str, Any]

Serialize decision to dict for logging/events.

ScopeEnvelope(dimensions: Mapping[str, str | None], attributes: Mapping[str, Any] = dict(), principal: PrincipalContext | None = None, metadata: Mapping[str, Any] = dict()) dataclass

Canonical, immutable scope context for a single operation.

dimensions holds normalised values keyed by canonical dimension name. A dimension key that is absent means "not specified". A dimension key present with a None value means "explicitly null" (e.g. workspace_id=None is the workspace-less bucket per decision D3).

Attributes:

Name Type Description
dimensions Mapping[str, str | None]

Mapping of canonical dimension key → normalised string (or None).
attributes Mapping[str, Any]

Non-dimension metadata (e.g. operation hints, flags).
principal PrincipalContext | None

Authenticated caller context; may be None for service calls.
metadata Mapping[str, Any]

Request-level tracing and observability metadata.
get_dim(key: str) -> str | None

Return the normalised value for key, or None if absent or null.

has_dim(key: str) -> bool

Return True if key is present (even with a None value).

to_dict(*, registry: DimensionRegistry | None = None, redact_sensitive: bool = False) -> dict[str, Any]

Serialise to a plain dict (suitable for JSON / logging).

Parameters:

Name Type Description Default
registry DimensionRegistry | None

Optional dimension registry used to determine sensitive dimensions for redaction.

 None
redact_sensitive bool

If True, replace sensitive dimension values with "<redacted>" when registry is provided.

 False
canonical_items(*, order: tuple[str, ...], sentinel: str | None) -> tuple[tuple[str, str | None], ...]

Return ordered key/value scope pairs with sentinel substitution.

This provides a canonical shape that can be transformed into any sink representation by simple joins/splits. Pass sentinel=None to keep null/absent values as None instead of substituting a filler.

canonical_segments(*, order: tuple[str, ...], sentinel: str) -> tuple[str, ...]

Return flat ordered segments: (k1, v1, k2, v2, ...).

from_dict(data: dict[str, Any], *, registry: DimensionRegistry | None = None) -> ScopeEnvelope classmethod

Construct from a plain dict, optionally resolving aliases via registry.

When registry is provided, aliases are resolved to canonical keys and values are normalised using each dimension's normalizer function. Without a registry, values are still validated via _default_normalizer (which rejects booleans, empty strings, and collections).

policy

Constraint framework for the unified scope system.

Constraints are evaluated in registration order. The first deny wins. All decisions are auditable via :class:ScopeDecision.

ScopeDecisionMetrics(total_allowed: int = 0, total_denied: int = 0, by_constraint: dict[str, int] = (lambda: defaultdict(int))(), by_reason_code: dict[str, int] = (lambda: defaultdict(int))(), by_enforcement_level: dict[str, int] = (lambda: defaultdict(int))()) dataclass

Aggregated metrics for scope decision outcomes.

Attributes:

Name Type Description
total_allowed int

Count of decisions with allowed=True.
total_denied int

Count of decisions with allowed=False.
by_constraint dict[str, int]

Breakdown of denials by constraint name.
by_reason_code dict[str, int]

Breakdown of denials by reason_code.
by_enforcement_level dict[str, int]

Breakdown of denials by enforcement_level.
to_dict() -> dict[str, Any]

Serialize metrics to dict.

ScopeMetricsCollector()

Thread-safe collector for scope decision metrics.

Can be used to track decisions across all ConstraintEvaluator instances for observability and debugging.

record_decision(decision: ScopeDecision, enforcement_level: str = 'log-only') -> None

Record a scope decision outcome.

get_metrics() -> ScopeDecisionMetrics

Get a snapshot of current metrics.

reset() -> None

Reset all metrics to zero.

ScopeConstraint

Bases: Protocol

Protocol that every scope constraint must implement.

Implementations should be stateless (or thread-safe) so they can be shared across requests.

evaluate(scope: ScopeEnvelope, context: dict[str, Any]) -> ScopeDecision

Evaluate the constraint against scope and context.

Parameters:

Name Type Description Default
scope ScopeEnvelope

The immutable scope envelope for the current operation.

 required
context dict[str, Any]

Runtime lookup context (e.g. DB session, service clients, request headers). Constraints must not mutate it.

 required

Returns:

Name Type Description
A ScopeDecision

class:~msflib.scope.model.ScopeDecision indicating allow/deny.

DimensionMatchConstraint(required_match: dict[str, str] | None = None, *, constraint_name: str | None = None)

Checks that dimension values in the scope match those in context.

Pass required_match as a mapping of {canonical_dim_key: context_key}. If the context key is present and the scope has that dimension, the values must match exactly. Missing context keys are skipped (not treated as deny).

Example::

constraint = DimensionMatchConstraint(
    required_match={"tenant_id": "current_tenant_id"},
)

If scope.get_dim("tenant_id") != context["current_tenant_id"], the constraint returns a deny decision.

MembershipConstraint(*, deny_without_principal: bool = True, fail_open: bool = False)

Stub: validates that the principal is a member of the scoped workspace.

Production implementations should inject a membership service or DB session via the context dict (key: membership_service).

By default (fail_open=False), missing or misconfigured services fail closed (deny) for security. Set fail_open=True to allow in shadow/log-only modes when the service is unavailable (e.g., for testing or during deployment).

If no principal is attached and the scope targets a concrete workspace, this constraint denies by default to avoid unauthenticated bypasses.

RoleConstraint(required_role: str, *, deny_without_principal: bool = True)

Stub: validates that the principal has the required role.

Requires the scope envelope to carry a PrincipalContext with a roles set. The required_role is checked against principal.roles.

If no principal is attached, behaviour depends on deny_without_principal: - True (default): deny. - False: allow (useful for service-to-service calls where role is implicit).

ChannelConstraint(required_channel: str, *, deny_without_channel: bool = True)

Validates that the request channel matches the required channel.

Channel is read from scope.attributes['channel'] first and falls back to context['current_channel'] when absent.

ConstraintEvaluator(constraints: list[ScopeConstraint] | None = None, enforcement_level: str = 'log-only')

Evaluates an ordered list of constraints and returns a composite decision.

Constraints are evaluated in order until the first deny. The enforcement_level controls what happens after that deny:

- ``log-only``: return *allow* (violations are logged but not enforced).
- ``shadow``: return *deny* (violations are logged; callers observe the
    denial without an exception being raised).
- ``enforce``: raise :class:`ConstraintViolationError` on the first deny.

Parameters:

Name Type Description Default
constraints list[ScopeConstraint] | None

Ordered list of :class:ScopeConstraint instances.

 None
enforcement_level str

One of "log-only", "shadow", "enforce". Defaults to "log-only" (D5 locked decision).

 'log-only'
last_decisions: list[ScopeDecision] property

Return decisions from the most recent :meth:evaluate call.

evaluate(scope: ScopeEnvelope, context: dict[str, Any] | None = None) -> ScopeDecision

Run all constraints in order; return first deny or final allow.

Side-effects: - Stores decisions in :attr:last_decisions for audit. - Logs violations at WARNING level. - Emits scope.decision events and records metrics for the first deny (including enforce raises) and for the final effective decision (an unconditional allow, a shadow deny, or a log-only override). A log-only override therefore emits and records both the original deny and the overridden allow.

The enforcement_level controls what happens on a deny: - log-only: Log and return allow. - shadow: Log and return deny (but do not raise). - enforce: Log and raise :class:ConstraintViolationError.

get_scope_metrics() -> ScopeDecisionMetrics

Get current scope decision metrics.

reset_scope_metrics() -> None

Reset metrics (useful for testing).

profiles

Operation profile registry and built-in profiles.

A profile defines the required/optional dimensions and policy-level enforcement settings for a named operation. The compiler validates envelopes against profiles before compiling.

ProfileRegistry(*, load_built_ins: bool = True)

Manages operation profiles used for scope validation.

Usage::

registry = ProfileRegistry()
registry.register(
    "my_op",
    {
        "required": ["tenant_id"],
        "required_nullable": ["workspace_id"],
        "optional": [],
    },
)
profile = registry.get("my_op")

Built-in profiles (retrieval.search, memory.put, etc.) are automatically loaded at construction time.

register(name: str, profile: dict[str, Any]) -> None

Register (or replace) a profile.

Parameters:

Name Type Description Default
name str

Operation name (e.g. "documents.ingest").

 required
profile dict[str, Any]

Dict with required, required_nullable, optional (and optionally constraints, enforcement_level keys).

 required

Raises:

Type Description
ValueError

If name is empty or profile is not a dict.
ProfileValidationError

If profile declares overlapping required and required_nullable dimensions.
get(name: str) -> dict[str, Any]

Return the profile for name.

Raises:

Type Description
ProfileValidationError

If name is not registered.
has(name: str) -> bool

Return True if name is registered.

list_all() -> list[str]

Return sorted list of registered profile names.

registry

Dimension registry for the unified scope system.

Scope dimensions are registered at application start-up. The registry can be locked (via :meth:lock) to prevent further registrations at runtime; locking is optional and caller-controlled.

DimensionRegistry()

Central registry of scope dimension specifications.

Usage::

registry = DimensionRegistry()
registry.register(DimensionSpec(canonical_key="project_id", priority=25))
registry.lock()

canonical = registry.resolve_alias("project_id")  # "project_id"
value = registry.normalize("project_id", 42)      # "42"

After :meth:lock is called, :meth:register raises :exc:RuntimeError.

register(spec: DimensionSpec) -> None

Register spec. Raises on duplicate canonical key or alias conflict.

get(key: str) -> DimensionSpec

Return the spec for key (canonical or alias).

Raises :exc:~msflib.scope.errors.UnknownDimensionError if not found.

resolve_alias(key: str) -> str

Return the canonical key for key (which may be canonical, alias, or key_alias).

Raises :exc:~msflib.scope.errors.UnknownDimensionError if unknown.

has(key: str) -> bool

Return True if key is a known canonical key, alias, or key_alias.

list_all() -> list[DimensionSpec]

Return all registered specs sorted by priority (highest first).

normalize(key: str, value: Any) -> str | None

Normalise value using the spec for key.

Returns None when value is None (explicit null).

validate(key: str, value: str | None) -> bool

Return True if value passes the spec's validator for key.

lock() -> None

Lock the registry. Further register() calls will raise.

is_locked() -> bool

Return True if the registry has been locked.

create_default_registry() -> DimensionRegistry

Create a new :class:DimensionRegistry pre-loaded with built-in dimensions.