Slotting Architecture · 12 min read

How to Implement Role-Based Access Control for Slotting Configs

An unscoped write to a velocity threshold or a fallback-routing rule does not fail loudly — it silently re-ranks thousands of SKUs and floods material handlers with relocation tasks on the next optimizer pass. This page solves one precise problem: how to gate every slotting-configuration write behind a role-and-scope check at the ingestion boundary, before the payload reaches the persistence layer or the constraint solver. It is a hands-on implementation companion to Enforcing Role- and Zone-Scoped Access Boundaries for Slotting, which frames the three policy axes; here we build the runnable Python that enforces them.

Prerequisites

Before wiring the decorator in, confirm you have:

  • Python 3.11+ — the code uses dataclasses, X | None unions, and datetime.now(timezone.utc).
  • A resolved actor identity — an authenticated principal carrying id, role, and zone_scope claims (from your API gateway, JWT, or service-mesh sidecar). RBAC assumes authentication already happened; it only authorizes.
  • A canonical role list agreed with operations — this guide uses slotting_admin, velocity_analyst, floor_operator, and audit_compliance.
  • A zone identifier on every config payload that resolves against the location graph in Location Hierarchy Mapping, so spatial scope can be checked, not assumed.
  • A durable checkpoint/version store (Postgres row version, or a Redis WATCH/MULTI key) for the optimistic-locking guard.
  • pytest>=8.0 for the validation checks in the last section.
The slotting-config RBAC decorator as a four-gate permission pipeline An authenticated actor carrying id, role, and zone_scope submits a config payload with a zone_id and version. The request passes left to right through four sequential gates: an action-scope gate checks the role's permitted operations in the RBAC matrix against the required operations; a spatial-scope gate requires the actor's zone_scope to equal the payload's zone_id; a TTL guard caps any temp_override at 3600 seconds and stamps an absolute expires_at; a version guard rejects any payload whose version is older than the current database version. A request that clears all four gates has an audit record injected and is granted through to the solver and persistence layer. Any gate that fails raises ScopeError or PermissionError before any write reaches the solver. An admin-approvable override the caller lacks is not hard-denied; it is staged to a Redis fallback queue where an elevated worker re-evaluates it. One write, four gates: action → zone → TTL → version, then audit & grant Actor + payload id · role · zone_scope config write: zone_id · version 1 · action scope 2 · spatial scope 3 · TTL guard 4 · version guard role ∈ matrix allows required_ops? actor.zone_scope == payload.zone_id? temp_override ttl ≤ 3600s → stamp expires_at payload.version ≥ db_version (optimistic lock) inject _audit actor · role ops · ts GRANT → solver / persistence DENY — raise ScopeError / PermissionError rejected at the ingestion boundary, before any write reaches the solver Redis fallback queue slotting:override:staging elevated worker re-approves admin-approvable override, not 403 grant path hard deny graceful fallback

Configuration Block

Externalize the role-to-operation matrix and the guard limits so operations can re-scope a role without a code deploy. The YAML below is the source of truth; the equivalent Python dict is what the enforcement layer loads at startup.

# slotting_rbac.yaml — one policy file per facility, version-controlled
rbac:
  matrix:                       # role -> set of permitted operations
    slotting_admin:   [create_zone, update_velocity, delete_slot, override_fallback]
    velocity_analyst: [read_velocity, update_velocity, export_logs]
    floor_operator:   [read_slot, temp_override]
    audit_compliance: [read_all, export_logs]
  guards:
    temp_override_max_ttl_s: 3600   # floor overrides self-expire within one shift-hour
    require_zone_scope: true        # deny cross-zone writes even if the role allows the action
    enforce_version: true           # reject stale-version payloads (optimistic lock)
  fallback:
    stream: "slotting:override:staging"  # Redis stream for admin-approval routing
    max_pending: 200                     # backpressure ceiling on the approval queue
# Equivalent Python config dict loaded at service start
SLOT_RBAC = {
    "matrix": {
        "slotting_admin":   {"create_zone", "update_velocity", "delete_slot", "override_fallback"},
        "velocity_analyst": {"read_velocity", "update_velocity", "export_logs"},
        "floor_operator":   {"read_slot", "temp_override"},
        "audit_compliance": {"read_all", "export_logs"},
    },
    "guards": {"temp_override_max_ttl_s": 3600, "require_zone_scope": True, "enforce_version": True},
    "fallback": {"stream": "slotting:override:staging", "max_pending": 200},
}

Each role maps to the operational reality of the SKU Velocity Taxonomy Design layer: a velocity_analyst may rewrite tier assignments (update_velocity) but never restructure zones, while a floor_operator may only stamp short-lived overrides that expire inside a shift.

Implementation

The core is a declarative decorator that resolves the actor against the matrix, checks zone scope, enforces the TTL and version guards, and injects an audit record — rejecting the request before func (the real persistence call) ever runs.

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from functools import wraps
from typing import Any, Callable

logger = logging.getLogger("slotting.rbac")


@dataclass(frozen=True)
class Actor:
    """Authenticated principal; authentication happens upstream, this only authorizes."""
    id: str
    role: str
    zone_scope: str | None = None


class ScopeError(PermissionError):
    """Raised when action-scope or spatial-scope resolution denies a request."""


def enforce_slotting_rbac(
    required_ops: set[str], cfg: dict[str, Any], zone_locked: bool = True
) -> Callable:
    """Gate a slotting-config write behind role, zone, TTL, and version checks."""
    def decorator(func: Callable) -> Callable:
        @wraps(func)  # preserve metadata so WMS API tracing keeps the real name
        def wrapper(actor: Actor, payload: dict[str, Any], *args, **kwargs):
            allowed = cfg["matrix"].get(actor.role, set())
            missing = required_ops - allowed
            if missing:
                logger.warning("RBAC DENY actor=%s role=%s missing=%s", actor.id, actor.role, missing)
                raise ScopeError(f"role '{actor.role}' lacks {missing}")

            if zone_locked and cfg["guards"]["require_zone_scope"]:
                if actor.zone_scope != payload.get("zone_id"):
                    logger.warning("SCOPE DENY actor=%s zone=%s", actor.id, payload.get("zone_id"))
                    raise ScopeError("zone-scope mismatch")

            if "temp_override" in required_ops:
                ttl = payload.get("ttl_seconds")
                if not ttl or ttl > cfg["guards"]["temp_override_max_ttl_s"]:
                    raise ValueError(f"temp_override needs ttl <= {cfg['guards']['temp_override_max_ttl_s']}s")
                payload["expires_at"] = datetime.now(timezone.utc).timestamp() + ttl

            if cfg["guards"]["enforce_version"] and "version" in payload:
                if payload["version"] < kwargs.get("db_version", payload["version"]):
                    raise ScopeError("stale payload version rejected (optimistic lock)")

            payload["_audit"] = {
                "actor_id": actor.id, "role": actor.role,
                "ops": sorted(required_ops),
                "ts": datetime.now(timezone.utc).isoformat(),
            }
            logger.info("RBAC GRANT actor=%s ops=%s zone=%s", actor.id, required_ops, payload.get("zone_id"))
            return func(actor, payload, *args, **kwargs)
        return wrapper
    return decorator

Step-by-Step Walkthrough

  1. Resolve the action scope. required_ops - allowed computes exactly which permissions the caller is missing against the matrix from config. An empty difference means every requested operation is permitted; anything left over raises ScopeError before any I/O. Declaring required_ops at the decorator site keeps the authorization contract next to the endpoint it protects.
  2. Check the spatial scope. When require_zone_scope is on, the actor’s zone_scope claim must equal the payload’s zone_id. This is what stops a floor_operator who owns ZONE_A from rewriting parameters in COLD_STORAGE, even though their role technically permits temp_override elsewhere.
  3. Bound temporary overrides in time. Any temp_override must carry a ttl_seconds at or under temp_override_max_ttl_s (3600s). The guard stamps expires_at so a background reaper can revert congestion hacks automatically — overrides never silently become permanent.
  4. Reject stale writes. With enforce_version on, a payload whose version is older than the current db_version is refused. This optimistic lock prevents a stale cached config from clobbering a concurrent peak-shift edit and corrupting live pick paths.
  5. Inject the audit trail. Every granted write carries a _audit record (actor, role, operations, UTC timestamp) so the audit_compliance role can export a machine-readable change log. @wraps keeps func.__name__ intact so this record ties back to the real handler in traces.

For an override_fallback the caller lacks (a floor_operator needing an admin-only change during an equipment failure), do not return a hard denial. Enqueue the payload to the Redis fallback.stream and let an elevated worker evaluate it against real-time pick-density before applying — a graceful degradation path instead of a blocked shift.

Verification

Assert the three invariants that matter — permitted ops pass, unscoped ops are denied, and an over-long TTL is rejected — before shipping. These run in the config service’s CI gate.

import pytest

CFG = SLOT_RBAC  # loaded from slotting_rbac.yaml


@enforce_slotting_rbac({"update_velocity"}, CFG)
def _write_velocity(actor: Actor, payload: dict) -> str:
    return "written"


def test_permitted_write_passes() -> None:
    analyst = Actor(id="u1", role="velocity_analyst", zone_scope="ZONE_A")
    assert _write_velocity(analyst, {"zone_id": "ZONE_A", "tier": "A"}) == "written"


def test_role_without_op_is_denied() -> None:
    operator = Actor(id="u2", role="floor_operator", zone_scope="ZONE_A")
    with pytest.raises(PermissionError):
        _write_velocity(operator, {"zone_id": "ZONE_A"})


def test_cross_zone_write_is_denied() -> None:
    analyst = Actor(id="u3", role="velocity_analyst", zone_scope="ZONE_A")
    with pytest.raises(PermissionError):
        _write_velocity(analyst, {"zone_id": "COLD_STORAGE", "tier": "B"})

A healthy run: test_permitted_write_passes returns "written" and logs RBAC GRANT actor=u1 ops={'update_velocity'} zone=ZONE_A, while both denial tests log an RBAC DENY / SCOPE DENY line and raise PermissionError. If the cross-zone test passes silently, require_zone_scope is off or the actor’s zone_scope claim is not populated — fix that before deploying, because it is the guard that contains blast radius.

Common Pitfalls

  • Enforcing at the presentation tier only. Hiding a button does not stop a direct API call. The decorator must sit at the config ingestion boundary (gateway or service entry), not in the UI, or malformed payloads still reach the solver.
  • Unbounded overrides. Accepting a temp_override without a TTL — or with a multi-day TTL — turns a congestion hotfix into permanent drift. Always cap ttl_seconds and run a reaper on expires_at.
  • Dropping the version guard under load. Peak-shift concurrency is exactly when two clients race the same zone config; without the optimistic-lock check a stale write wins and re-ranks a zone against yesterday’s velocity.
  • Hard-denying fallback-eligible requests. Returning 403 to a floor_operator during an equipment failure blocks the shift instead of routing to the staging stream. Reserve hard denials for genuine scope violations, not time-sensitive escalations.

FAQ

Where in the stack should the RBAC decorator live?

At the configuration ingestion boundary — the API gateway, service-mesh entry, or the first handler of the slotting config service — never in the presentation tier. Enforcing it upstream guarantees that an unauthorized or malformed payload is rejected before it can touch the persistence layer or the constraint solver, which is the only place the guarantee actually holds.

How is zone scope different from role scope?

Role scope answers what action a caller may perform (read, propose, override, write); zone scope answers where they may perform it. They are orthogonal: a velocity_analyst may hold update_velocity globally, but the spatial-scope check still refuses the write if the payload’s zone_id is outside their zone_scope. Both must pass, and both resolve against the location graph in Location Hierarchy Mapping.

Why route some denials to a fallback queue instead of returning 403?

Because warehouse operations cannot halt for an authorization gap during an equipment failure or congestion event. A floor_operator who needs an admin-only override gets their payload staged to a Redis stream where an elevated worker evaluates it against live pick-density metrics, rather than a blocked shift. Genuine scope violations still fail hard; only time-sensitive, admin-approvable actions degrade gracefully.

How do TTL-bound overrides prevent configuration drift?

Every temp_override must carry a ttl_seconds at or under 3600, and the guard stamps an absolute expires_at. A background reaper reverts the override once that timestamp passes, so a real-time congestion hack can never silently become the permanent slotting rule that quietly degrades pick paths.