add allow and deny mode sequence enforcement

Author: artoodavid

.github/workflows/ci.yml

@@ -16,10 +16,10 @@ jobs:
 
     steps:
       - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4
 
       - name: Setup Python
-        uses: actions/setup-python@v5
+        uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5
         with:
           python-version: ${{ matrix.python-version }}
           cache: 'pip'
@@ -45,10 +45,10 @@ jobs:
     continue-on-error: true
     steps:
       - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4
 
       - name: Setup Python
-        uses: actions/setup-python@v5
+        uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5
         with:
           python-version: '3.11'
           cache: 'pip'
@@ -72,10 +72,10 @@ jobs:
     needs: [ test ]
     steps:
       - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4
 
       - name: Setup Python
-        uses: actions/setup-python@v5
+        uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5
         with:
           python-version: '3.11'
           cache: 'pip'
@@ -95,7 +95,7 @@ jobs:
           python -m build
 
       - name: Upload dist artifacts
-        uses: actions/upload-artifact@v4
+        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4
         with:
           name: dist
-          path: dist/* 
+          path: dist/*

.github/workflows/release.yml

@@ -14,10 +14,10 @@ jobs:
 
     steps:
       - name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4
 
       - name: Setup Python
-        uses: actions/setup-python@v5
+        uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5
         with:
           python-version: '3.11'
           cache: 'pip'
@@ -42,8 +42,8 @@ jobs:
 
       - name: Publish to PyPI
         if: startsWith(github.ref, 'refs/tags/v')
-        uses: pypa/gh-action-pypi-publish@release/v1
-        with:
-          user: __token__
-          password: ${{ secrets.PYPI_API_TOKEN }}
-          skip-existing: true 
+        env:
+          TWINE_USERNAME: __token__
+          TWINE_PASSWORD: ${{ secrets.PYPI_API_TOKEN }}
+        run: |
+          twine upload --skip-existing dist/*

d2/__init__.py

@@ -5,12 +5,12 @@
 """
 Artoo SDK: A client-side library for enforcing RBAC on LLM tools.
 
-Public API is stable as of 1.2.0. Backward-incompatible changes will only occur
+Public API is stable as of 1.2.1. Backward-incompatible changes will only occur
 in a new major version. Deprecated symbols (if any) will be retained for at least
 one minor version and warned via release notes before removal.
 """
 
-__version__ = "1.2.0"
+__version__ = "1.2.1"
 
 # Export the simple setter alongside the context-manager variant
 from .context import (

d2/decorator.py

@@ -239,10 +239,12 @@ async def check_and_run():
                         user_context = get_user_context()
 
                         if not is_allowed:
+                            user_roles = list(user_context.roles) if user_context and user_context.roles else []
                             error = PermissionDeniedError(
                                 tool_id=effective_tool_id,
                                 user_id=user_context.user_id if user_context else "unknown",
                                 roles=user_context.roles if user_context else [],
+                                reason=f"rbac_denied: role_missing (present roles: {user_roles})",
                             )
                             # Record a denied tool invocation outcome for observability
                             tool_invocation_total.add(1, {"tool_id": effective_tool_id, "status": "denied"})
@@ -264,7 +266,7 @@ async def check_and_run():
                             return await _handle_permission_denied(validation_error)
 
                         # Layer 3: Sequence enforcement
-                        sequence_rules = await manager.get_sequence_rules()
+                        sequence_mode, sequence_rules = await manager.get_sequence_rules()
                         if sequence_rules:
                             sequence_start = time.perf_counter()
                             from .runtime.sequence import SequenceValidator
@@ -275,7 +277,8 @@ async def check_and_run():
                             sequence_error = validator.validate_sequence(
                                 current_history=user_context.call_history if user_context else (),
                                 next_tool_id=effective_tool_id,
-                                sequence_rules=sequence_rules
+                                sequence_rules=sequence_rules,
+                                mode=sequence_mode
                             )
                             sequence_duration_ms = (time.perf_counter() - sequence_start) * 1000.0
                             guardrail_latency_ms.record(sequence_duration_ms, {
@@ -669,10 +672,12 @@ async def async_wrapper(*args, **kwargs):
                     user_context = get_user_context()
 
                     if not is_allowed:
+                        user_roles = list(user_context.roles) if user_context and user_context.roles else []
                         error = PermissionDeniedError(
                             tool_id=effective_tool_id,
                             user_id=user_context.user_id if user_context else "unknown",
                             roles=user_context.roles if user_context else [],
+                            reason=f"rbac_denied: role_missing (present roles: {user_roles})",
                         )
                         # Record a denied tool invocation outcome for observability
                         tool_invocation_total.add(1, {"tool_id": effective_tool_id, "status": "denied"})
@@ -693,7 +698,7 @@ async def async_wrapper(*args, **kwargs):
                         return await _handle_permission_denied(validation_error)
 
                     # Layer 3: Sequence enforcement
-                    sequence_rules = await manager.get_sequence_rules()
+                    sequence_mode, sequence_rules = await manager.get_sequence_rules()
                     if sequence_rules:
                         sequence_start = time.perf_counter()
                         from .runtime.sequence import SequenceValidator
@@ -704,7 +709,8 @@ async def async_wrapper(*args, **kwargs):
                         sequence_error = validator.validate_sequence(
                             current_history=user_context.call_history if user_context else (),
                             next_tool_id=effective_tool_id,
-                            sequence_rules=sequence_rules
+                            sequence_rules=sequence_rules,
+                            mode=sequence_mode
                         )
                         sequence_duration_ms = (time.perf_counter() - sequence_start) * 1000.0
                         guardrail_latency_ms.record(sequence_duration_ms, {

d2/policy/bundle.py

@@ -29,7 +29,8 @@ class PolicyBundle:
     etag: Optional[str] = None  # For analytics and caching
     tool_to_roles: Dict[str, Set[str]] = field(default_factory=dict, repr=False)
     tool_conditions: Dict[str, List[Dict[str, Any]]] = field(default_factory=dict, repr=False)
-    role_to_sequences: Dict[str, List[Dict[str, Any]]] = field(default_factory=dict, repr=False)
+    # Map role -> SequenceRules object (holds rules list + mode)
+    role_to_sequences: Dict[str, Any] = field(default_factory=dict, repr=False)
     # Data flow tracking: which labels each tool produces
     tool_labels: Dict[str, Set[str]] = field(default_factory=dict, repr=False)
     # Data flow tracking: which labels block which tools
@@ -119,14 +120,31 @@ def __post_init__(self):
                         )
 
                 # Parse sequence rules (apply to this specific role)
-                sequence_rules = policy.get("sequence", [])
-                if sequence_rules:
-                    logger.debug("Processing %d sequence rules for role '%s'", len(sequence_rules), role)
+                sequence_raw = policy.get("sequence", [])
+                
+                # Handle both legacy list format and new nested format
+                if isinstance(sequence_raw, dict):
+                    # New format: {"mode": "deny", "rules": [...]}
+                    mode = sequence_raw.get("mode")
+                    rules_list = sequence_raw.get("rules", [])
+                else:
+                    # Legacy format: list of rules directly, default to allow mode
+                    mode = "allow"
+                    rules_list = sequence_raw if isinstance(sequence_raw, list) else []
+
+                # Store sequence rules if they exist OR if mode is "deny" (empty deny = block all)
+                if rules_list or (mode == "deny"):
+                    logger.debug("Processing %d sequence rules for role '%s' (mode=%s)", len(rules_list), role, mode)
                     # Validate @group references without expanding (lazy expansion at runtime)
                     tool_groups = self._extract_tool_groups(policy_content)
-                    validated_rules = self._validate_sequence_rules(sequence_rules, tool_groups)
-                    self.role_to_sequences[role] = validated_rules
-                    logger.debug("Validated %d sequence rules for role '%s' (lazy @group expansion)", len(validated_rules), role)
+                    validated_rules = self._validate_sequence_rules(rules_list, tool_groups)
+                    
+                    # Store as simple dict to avoid pickling issues with dataclasses
+                    self.role_to_sequences[role] = {
+                        "mode": mode,
+                        "rules": validated_rules
+                    }
+                    logger.debug("Validated %d sequence rules for role '%s'", len(validated_rules), role)
 
         logger.debug("Final tool_to_roles mapping: %s", dict(self.tool_to_roles))
         logger.debug("Final role_to_sequences mapping: %s", dict(self.role_to_sequences))
@@ -140,16 +158,17 @@ def all_known_tools(self) -> Set[str]:
 
         return set(self.tool_to_roles.keys())
 
-    def get_sequence_rules(self, role: str) -> List[Dict[str, Any]]:
-        """Get sequence rules for a specific role.
+    def get_sequence_rules(self, role: str) -> Dict[str, Any]:
+        """Get sequence configuration for a specific role.
         
         Args:
             role: Role name
             
         Returns:
-            List of sequence rules (may contain @group references for lazy expansion)
+            Dict with 'mode' (str) and 'rules' (List[Dict]) keys.
+            Returns None if no sequence rules defined for role.
         """
-        return self.role_to_sequences.get(role, [])
+        return self.role_to_sequences.get(role)
 
     def get_tool_groups(self) -> Dict[str, List[str]]:
         """Get tool_groups from policy metadata for lazy sequence expansion.

d2/policy/manager.py

@@ -579,38 +579,52 @@ def _collect_conditions(tool_identifier: str):
 
         return applicable
 
-    async def get_sequence_rules(self) -> List[Dict[str, Any]]:
-        """Retrieve sequence rules for the current user's roles.
+    async def get_sequence_rules(self) -> tuple[Optional[str], List[Dict[str, Any]]]:
+        """Retrieve sequence rules and mode for the current user's roles.
         
         Returns:
-            List of sequence rule dictionaries (deny patterns with reasons)
+            Tuple of (mode, rules_list)
+            Mode is "deny" if any role uses deny mode, otherwise "allow" (or None if no rules).
             
         Example:
-            [
-                {"deny": ["database.read", "web.request"], "reason": "Exfiltration"},
-                {"deny": ["secrets.get", "web.request"], "reason": "Secret leak"}
-            ]
+            ("allow", [{"deny": ["database.read", "web.request"], "reason": "Exfiltration"}])
         """
         await self._init_complete.wait()
         bundle = self._get_bundle()
         user_context = get_user_context()
 
         if user_context is None or not user_context.roles:
-            return []
+            return None, []
 
         user_roles = set(user_context.roles)
 
         # Admin wildcard role bypasses sequence enforcement
         if "*" in user_roles:
-            return []
+            return None, []
 
-        # Collect all sequence rules for user's roles
+        # Collect all sequence rules and determine effective mode
         all_rules: List[Dict[str, Any]] = []
+        effective_mode = "allow"
+        has_rules = False
+
         for role in user_roles:
-            role_rules = bundle.role_to_sequences.get(role, [])
-            all_rules.extend(role_rules)
+            sequence_data = bundle.get_sequence_rules(role)
+            if sequence_data:
+                has_rules = True
+                role_mode = sequence_data.get("mode", "allow")
+                role_rules = sequence_data.get("rules", [])
+                
+                # If any role uses 'deny' mode (allowlist), the whole check 
+                # shifts to 'deny' mode for maximum security (least privilege).
+                if role_mode.lower() == "deny":
+                    effective_mode = "deny"
+                
+                all_rules.extend(role_rules)
+
+        if not has_rules:
+            return None, []
 
-        return all_rules
+        return effective_mode, all_rules
 
     async def is_tool_in_policy_async(self, tool_id: str) -> bool:
         """Async: Checks if a tool ID is defined in the policy at all."""

d2/runtime/sequence.py

@@ -127,7 +127,27 @@ def validate_sequence(
                 continue
 
             # Check if the proposed sequence matches the pattern
-            if self._matches_pattern(proposed_sequence, pattern, effective_mode):
+            # Check for match
+            match = False
+            
+            if effective_mode == "deny" and is_allow:
+                # In deny mode, we need strict control to prevent arbitrary extensions.
+                # A sequence is allowed if:
+                # 1. It is a valid prefix of an allowed pattern (progressing step-by-step)
+                #    e.g. [A] matches start of [A, B]
+                if self._is_prefix_match(proposed_sequence, pattern):
+                    match = True
+                # 2. It completes an allowed pattern (allowing gaps/subsequences)
+                #    But it MUST end with the pattern's last element to prevent extension
+                #    e.g. [A, ... B] is allowed, but [A, B, ... C] is not
+                elif self._matches_pattern(proposed_sequence, pattern, effective_mode):
+                    if self._tool_matches_element(next_tool_id, pattern[-1]):
+                        match = True
+            else:
+                # In allow mode (or deny rules), use standard subsequence matching (supports gaps)
+                match = self._matches_pattern(proposed_sequence, pattern, effective_mode)
+
+            if match:
                 if is_allow:
                     # Allow rule matches
                     has_matching_allow = True
@@ -168,6 +188,19 @@ def validate_sequence(
                 )
 
             return None
+            
+    def _is_prefix_match(self, sequence: list[str], pattern: list[str]) -> bool:
+        """Check if sequence is a valid prefix of the pattern (strict order, no gaps).
+        
+        Used in deny mode to allow partial progress through an allowed sequence.
+        """
+        if len(sequence) > len(pattern):
+            return False
+        
+        for i, tool in enumerate(sequence):
+            if not self._tool_matches_element(tool, pattern[i]):
+                return False
+        return True
 
     def _matches_pattern(
         self, 

examples/data_flow_demo.py

@@ -381,3 +381,4 @@ async def main():
 if __name__ == "__main__":
     asyncio.run(main())
 
+

examples/sequence_modes_demo.py

@@ -295,3 +295,4 @@ async def main():
 
 
 
+

examples/sequence_modes_policy.yaml

@@ -4,7 +4,7 @@
 metadata:
   name: "sequence-modes-demo"
   description: "Demonstrates allow mode (blocklist) and deny mode (allowlist) for sequences"
-  expires: "2026-12-31T23:59:59+00:00"
+  expires: "2026-12-26T23:59:59+00:00"
 
   # Define tool groups for use in sequence rules
   tool_groups:
@@ -311,3 +311,4 @@ policies:
 
 
 
+