From 92c8f2f03bd1de01eede4d41e6e3a9a8965d4301 Mon Sep 17 00:00:00 2001
From: hataa <79907651+hata33@users.noreply.github.com>
Date: Tue, 21 Jul 2026 09:23:14 +0800
Subject: [PATCH] feat(authz): add built-in RBAC provider and provider factory
 (#4260)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* feat(authz): add built-in RBAC provider and provider factory (Phase 1A-2, #4063)

Phase 1A-2: RBAC provider + provider factory. No runtime behavior change.

New authz/rbac.py — RbacAuthorizationProvider:
- allow: '*' / True / list / [] / missing → deny-wins semantics
- deny always overrides all allow forms
- resource name explicit mapping (tool→tools, model→models, etc.)
- unknown/missing role raises ValueError (never silent allow)
- config compiled to immutable frozensets at construction
- filter_resources preserves order, no mutation, consistent with authorize
- sync == async decisions

New authz/runtime.py — resolve_authorization_provider:
- disabled → None (no import attempted)
- enabled + no provider → ValueError
- invalid class path / construction failure → ValueError with path
- isinstance Protocol check post-construction
- no caching, no fail_closed/default_role injection

48 tests (37 RBAC + 11 factory). No config schema change, no config_version bump.
Per RFC #4063 Phase 1A-2. Layer 1/Layer 2 wiring deferred to Phase 1B.

* fix(authz): reject unknown RBAC provider config

Fail fast on misspelled top-level RBAC settings, cover factory error propagation, and record Phase 1B policy and audit caveats.

* fix(authz): reject unreachable resource aliases

Fail fast when RBAC config uses reserved request-side aliases, preserve same-name and custom resources, and add regression coverage for every mapped alias.

* fix(authz): validate RBAC request identifiers
---
 .../harness/deerflow/authz/__init__.py        |   4 +
 .../packages/harness/deerflow/authz/rbac.py   | 254 +++++++++++
 .../harness/deerflow/authz/runtime.py         |  50 +++
 backend/tests/test_authorization_runtime.py   | 211 +++++++++
 .../tests/test_rbac_authorization_provider.py | 416 ++++++++++++++++++
 ...able-authorization-implementation-notes.md |  48 ++
 6 files changed, 983 insertions(+)
 create mode 100644 backend/packages/harness/deerflow/authz/rbac.py
 create mode 100644 backend/packages/harness/deerflow/authz/runtime.py
 create mode 100644 backend/tests/test_authorization_runtime.py
 create mode 100644 backend/tests/test_rbac_authorization_provider.py

diff --git a/backend/packages/harness/deerflow/authz/__init__.py b/backend/packages/harness/deerflow/authz/__init__.py
index b7cdcd1b8..b0218e620 100644
--- a/backend/packages/harness/deerflow/authz/__init__.py
+++ b/backend/packages/harness/deerflow/authz/__init__.py
@@ -3,6 +3,8 @@
 from deerflow.authz.adapter import GuardrailAuthorizationAdapter
 from deerflow.authz.principal import build_principal_from_context, normalize_authz_attributes
 from deerflow.authz.provider import AuthorizationProvider, AuthzDecision, AuthzReason, AuthzRequest, Principal
+from deerflow.authz.rbac import RbacAuthorizationProvider
+from deerflow.authz.runtime import resolve_authorization_provider
 
 __all__ = [
     "AuthzDecision",
@@ -11,6 +13,8 @@ __all__ = [
     "AuthorizationProvider",
     "GuardrailAuthorizationAdapter",
     "Principal",
+    "RbacAuthorizationProvider",
     "build_principal_from_context",
     "normalize_authz_attributes",
+    "resolve_authorization_provider",
 ]
diff --git a/backend/packages/harness/deerflow/authz/rbac.py b/backend/packages/harness/deerflow/authz/rbac.py
new file mode 100644
index 000000000..ff315b2d3
--- /dev/null
+++ b/backend/packages/harness/deerflow/authz/rbac.py
@@ -0,0 +1,254 @@
+"""Built-in RBAC authorization provider.
+
+Reads a role→resource policy from config and compiles it into immutable
+structures at construction time. Deny always wins over allow. Unknown or
+missing roles raise ``ValueError`` (not a silent allow) so that the execution
+layer's ``fail_closed`` can make the final decision.
+
+See ``docs/plans/2026-07-15-authz-phase1a-implementation-plan.md`` §3.3 for
+the full semantic table.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from deerflow.authz.provider import (
+    AuthzDecision,
+    AuthzReason,
+    AuthzRequest,
+    Principal,
+)
+
+# Explicit resource-type → config-key mapping. Prevents silent mis-lookup
+# when ``AuthzRequest.resource`` (singular, e.g. "tool") doesn't match the
+# config key (plural, e.g. "tools").
+_RESOURCE_POLICY_KEYS: dict[str, str] = {
+    "tool": "tools",
+    "model": "models",
+    "skill": "skills",
+    "sandbox": "sandbox",
+    "mcp_server": "mcp_servers",
+    "route": "routes",
+}
+
+_ALL = object()  # sentinel meaning "allow all candidates"
+_ABSENT = object()  # sentinel meaning "key not present in dict"
+
+
+# The only supported keys in a resource policy dict. Any other key (typos,
+# unknown fields) is rejected at construction to prevent silent mis-grants.
+_SUPPORTED_POLICY_KEYS: frozenset[str] = frozenset({"allow", "deny"})
+
+
+def _require_non_empty_string(value: object, *, field: str) -> str:
+    """Return a validated request identifier or raise a stable boundary error."""
+    if not isinstance(value, str) or not value:
+        raise ValueError(f"{field} must be a non-empty string, got {value!r}")
+    return value
+
+
+class _CompiledPolicy:
+    """Immutable, pre-validated policy for a single (role, resource_type) pair."""
+
+    __slots__ = ("allowed", "denied")
+
+    def __init__(self, *, allowed: frozenset[str] | object, denied: frozenset[str]):
+        self.allowed = allowed
+        self.denied = denied
+
+    def is_allowed(self, target: str) -> bool:
+        # Deny always wins.
+        if target in self.denied:
+            return False
+        if self.allowed is _ALL:
+            return True
+        return target in self.allowed
+
+
+class RbacAuthorizationProvider:
+    """Built-in role-based authorization provider.
+
+    Configured via ``roles`` mapping where each role maps resource-type keys
+    to ``{allow: ..., deny: [...]}`` policies. Policy configuration is fully
+    validated at construction; the request path validates identifiers before
+    performing membership checks.
+
+    Policies are scoped by role, resource, and target. ``AuthzRequest.action``
+    is accepted for protocol compatibility but is not a rule dimension in this
+    built-in provider.
+
+    Example config::
+
+        roles:
+          admin:
+            tools: {allow: "*"}
+          user:
+            tools: {allow: "*", deny: ["update_agent"]}
+          guest:
+            tools: {allow: ["web_search", "read_file"]}
+    """
+
+    name = "rbac"
+
+    def __init__(self, *, roles: dict[str, Any] | object = _ABSENT, **kwargs: Any) -> None:
+        if kwargs:
+            raise ValueError(f"unknown provider config keys {sorted(kwargs, key=repr)}; supported: ['roles']")
+        if roles is _ABSENT:
+            raise ValueError("missing required provider config key 'roles'")
+        if not isinstance(roles, dict):
+            raise ValueError(f"roles must be a dict, got {type(roles).__name__}")
+
+        # Compile all policies up front.
+        self._policies: dict[tuple[str, str], _CompiledPolicy] = {}
+        self._known_roles: frozenset[str] = frozenset(roles.keys())
+
+        for role_name, role_config in roles.items():
+            if not isinstance(role_name, str) or not role_name:
+                raise ValueError(f"role name must be a non-empty string, got {role_name!r}")
+            if not isinstance(role_config, dict):
+                raise ValueError(f"role '{role_name}' config must be a dict, got {type(role_config).__name__}")
+
+            for resource_key, resource_policy in role_config.items():
+                if not isinstance(resource_key, str) or not resource_key:
+                    raise ValueError(f"role '{role_name}' has invalid resource key {resource_key!r}")
+                mapped_resource_key = _RESOURCE_POLICY_KEYS.get(resource_key)
+                if mapped_resource_key is not None and mapped_resource_key != resource_key:
+                    raise ValueError(f"role '{role_name}' resource key '{resource_key}' is a reserved request alias; use '{mapped_resource_key}' in RBAC config")
+                if not isinstance(resource_policy, dict):
+                    raise ValueError(f"role '{role_name}' resource '{resource_key}' must be a dict, got {type(resource_policy).__name__}")
+
+                compiled = self._compile_resource_policy(role_name, resource_key, resource_policy)
+                self._policies[(role_name, resource_key)] = compiled
+
+    @staticmethod
+    def _compile_resource_policy(
+        role_name: str,
+        resource_key: str,
+        policy: dict[str, Any],
+    ) -> _CompiledPolicy:
+        """Validate and compile a single resource policy into immutable structures.
+
+        Distinguishes "key absent" (use default) from "key present but null"
+        (invalid — reject). Unknown keys (typos) are rejected to prevent
+        silent mis-grants.
+        """
+        # --- reject unknown keys (catch typos like "alow") ---
+        unknown_keys = set(policy.keys()) - _SUPPORTED_POLICY_KEYS
+        if unknown_keys:
+            raise ValueError(f"role '{role_name}' resource '{resource_key}': unknown policy keys {sorted(unknown_keys, key=repr)}; supported: {sorted(_SUPPORTED_POLICY_KEYS)}")
+
+        # --- allow ---
+        raw_allow = policy.get("allow", _ABSENT)
+        if raw_allow is _ABSENT:
+            allowed: frozenset[str] | object = _ALL  # missing allow = allow all (deny still applies)
+        elif raw_allow is None:
+            raise ValueError(f"role '{role_name}' resource '{resource_key}': allow must not be null; omit the key, or use '*' / bool / list")
+        elif raw_allow is True:
+            allowed = _ALL
+        elif raw_allow is False:
+            allowed = frozenset()  # allow: false = deny all
+        elif isinstance(raw_allow, str):
+            if raw_allow == "*":
+                allowed = _ALL
+            else:
+                raise ValueError(f"role '{role_name}' resource '{resource_key}': allow string must be '*', got {raw_allow!r}")
+        elif isinstance(raw_allow, (list, tuple)):
+            for item in raw_allow:
+                if not isinstance(item, str) or not item:
+                    raise ValueError(f"role '{role_name}' resource '{resource_key}': allow list contains non-string or empty item {item!r}")
+            allowed = frozenset(raw_allow)
+        else:
+            raise ValueError(f"role '{role_name}' resource '{resource_key}': allow must be '*', bool, or list of strings, got {type(raw_allow).__name__}")
+
+        # --- deny ---
+        raw_deny = policy.get("deny", _ABSENT)
+        if raw_deny is _ABSENT or raw_deny is None:
+            if raw_deny is None:
+                raise ValueError(f"role '{role_name}' resource '{resource_key}': deny must not be null; omit the key for no deny list")
+            denied: frozenset[str] = frozenset()
+        elif isinstance(raw_deny, (list, tuple)):
+            for item in raw_deny:
+                if not isinstance(item, str) or not item:
+                    raise ValueError(f"role '{role_name}' resource '{resource_key}': deny list contains non-string or empty item {item!r}")
+            denied = frozenset(raw_deny)
+        else:
+            raise ValueError(f"role '{role_name}' resource '{resource_key}': deny must be a list of strings, got {type(raw_deny).__name__}")
+
+        return _CompiledPolicy(allowed=allowed, denied=denied)
+
+    def _resolve_policy(
+        self,
+        principal: Principal,
+        resource: str,
+        *,
+        resource_field: str = "resource",
+    ) -> _CompiledPolicy | None:
+        """Look up the compiled policy for (role, resource_type).
+
+        Returns ``None`` if no policy is configured for this role+resource
+        (meaning: unrestricted). Raises ``ValueError`` for invalid resource
+        identifiers and unknown or missing roles.
+        """
+        role = principal.role
+        if role is None or role == "":
+            raise ValueError("Principal has no role; cannot evaluate RBAC policy")
+
+        if role not in self._known_roles:
+            raise ValueError(f"Unknown role '{role}'; known roles: {sorted(self._known_roles)}")
+
+        resource = _require_non_empty_string(resource, field=resource_field)
+        resource_key = _RESOURCE_POLICY_KEYS.get(resource, resource)
+        return self._policies.get((role, resource_key))
+
+    def authorize(self, request: AuthzRequest) -> AuthzDecision:
+        """Evaluate a single authorization request."""
+        policy = self._resolve_policy(request.principal, request.resource)
+        target = _require_non_empty_string(request.target, field="target")
+        if policy is None:
+            # No policy for this role+resource → unrestricted.
+            return AuthzDecision(
+                allow=True,
+                reasons=[AuthzReason(code="authz.no_policy", message="no policy configured")],
+                policy_id="rbac:unrestricted",
+            )
+
+        if policy.is_allowed(target):
+            return AuthzDecision(
+                allow=True,
+                reasons=[AuthzReason(code="authz.allowed")],
+                policy_id="rbac:allow",
+            )
+        return AuthzDecision(
+            allow=False,
+            reasons=[
+                AuthzReason(
+                    code="authz.denied",
+                    message=f"role '{request.principal.role}' is denied '{target}' on resource '{request.resource}'",
+                )
+            ],
+            policy_id="rbac:deny",
+        )
+
+    async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
+        return self.authorize(request)
+
+    def filter_resources(
+        self,
+        principal: Principal,
+        resource_type: str,
+        candidates: list[str],
+    ) -> list[str]:
+        """Batch visibility filter.
+
+        Preserves candidate order and duplicates, never adds items, and raises
+        the same role/resource errors as :meth:`authorize`.
+        """
+        policy = self._resolve_policy(principal, resource_type, resource_field="resource_type")
+        if not isinstance(candidates, list):
+            raise ValueError(f"candidates must be a list, got {type(candidates).__name__}")
+        validated_candidates = [_require_non_empty_string(candidate, field=f"candidates[{index}]") for index, candidate in enumerate(candidates)]
+        if policy is None:
+            return validated_candidates
+
+        return [candidate for candidate in validated_candidates if policy.is_allowed(candidate)]
diff --git a/backend/packages/harness/deerflow/authz/runtime.py b/backend/packages/harness/deerflow/authz/runtime.py
new file mode 100644
index 000000000..8fb4062d7
--- /dev/null
+++ b/backend/packages/harness/deerflow/authz/runtime.py
@@ -0,0 +1,50 @@
+"""Provider factory — resolves and constructs the configured AuthorizationProvider.
+
+This is the single entry point for creating an authorization provider from
+``AuthorizationConfig``. It does not cache instances (Phase 1B resolves once
+per agent build and passes the same instance to Layer 1 and Layer 2).
+"""
+
+from __future__ import annotations
+
+from deerflow.authz.provider import AuthorizationProvider
+from deerflow.config.authorization_config import AuthorizationConfig
+from deerflow.reflection import resolve_variable
+
+
+def resolve_authorization_provider(
+    config: AuthorizationConfig,
+) -> AuthorizationProvider | None:
+    """Resolve the authorization provider from config.
+
+    Returns:
+        A constructed ``AuthorizationProvider`` instance, or ``None`` if
+        authorization is disabled.
+
+    Raises:
+        ValueError: If ``enabled`` is True but no provider is configured,
+            or if the class path is invalid / construction fails / the
+            instance does not satisfy the ``AuthorizationProvider`` Protocol.
+    """
+    if not config.enabled:
+        return None
+
+    if config.provider is None:
+        raise ValueError("authorization.enabled is true but no provider is configured; set authorization.provider.use to a class path")
+
+    class_path = config.provider.use
+    try:
+        provider_cls = resolve_variable(class_path, expected_type=type)
+    except (ImportError, ValueError) as err:
+        raise ValueError(f"Failed to resolve authorization provider class '{class_path}': {err}") from err
+
+    kwargs = dict(config.provider.config) if config.provider.config else {}
+    try:
+        instance = provider_cls(**kwargs)
+    except Exception as err:
+        raise ValueError(f"Failed to construct authorization provider '{class_path}': {err}") from err
+
+    if not isinstance(instance, AuthorizationProvider):
+        raise ValueError(f"Authorization provider '{class_path}' does not satisfy the AuthorizationProvider Protocol")
+
+    return instance
diff --git a/backend/tests/test_authorization_runtime.py b/backend/tests/test_authorization_runtime.py
new file mode 100644
index 000000000..159bdf0e6
--- /dev/null
+++ b/backend/tests/test_authorization_runtime.py
@@ -0,0 +1,211 @@
+"""Tests for resolve_authorization_provider — the provider factory."""
+
+from __future__ import annotations
+
+import pytest
+
+from deerflow.authz.provider import AuthorizationProvider
+from deerflow.authz.runtime import resolve_authorization_provider
+from deerflow.config.authorization_config import AuthorizationConfig, AuthorizationProviderConfig
+
+
+class TestDisabled:
+    """enabled=False must return None without importing anything."""
+
+    def test_disabled_returns_none(self):
+        config = AuthorizationConfig(enabled=False)
+        assert resolve_authorization_provider(config) is None
+
+    def test_disabled_with_provider_config_returns_none(self):
+        """Even with a provider configured, disabled means None."""
+        config = AuthorizationConfig(
+            enabled=False,
+            provider=AuthorizationProviderConfig(use="nonexistent.module:FakeProvider"),
+        )
+        assert resolve_authorization_provider(config) is None
+
+
+class TestMissingProvider:
+    """enabled=True but no provider configured must fail clearly."""
+
+    def test_enabled_without_provider_raises(self):
+        config = AuthorizationConfig(enabled=True)
+        with pytest.raises(ValueError, match="no provider is configured"):
+            resolve_authorization_provider(config)
+
+
+class TestValidProvider:
+    """Built-in and custom providers resolve through the same path."""
+
+    def test_builtin_rbac_resolves(self):
+        config = AuthorizationConfig(
+            enabled=True,
+            provider=AuthorizationProviderConfig(
+                use="deerflow.authz.rbac:RbacAuthorizationProvider",
+                config={"roles": {"admin": {"tools": {"allow": "*"}}}},
+            ),
+        )
+        provider = resolve_authorization_provider(config)
+        assert provider is not None
+        assert isinstance(provider, AuthorizationProvider)
+        assert provider.name == "rbac"
+
+    def test_custom_provider_resolves(self):
+        """A provider defined in an importable module resolves through the same path."""
+        config = AuthorizationConfig(
+            enabled=True,
+            provider=AuthorizationProviderConfig(
+                use="deerflow.authz.rbac:RbacAuthorizationProvider",
+                config={"roles": {"user": {"tools": {"allow": ["web_search"]}}}},
+            ),
+        )
+        provider = resolve_authorization_provider(config)
+        assert provider is not None
+
+
+class TestInvalidClassPath:
+    """Invalid class paths must raise with the path in the error."""
+
+    def test_nonexistent_module_raises_with_path(self):
+        config = AuthorizationConfig(
+            enabled=True,
+            provider=AuthorizationProviderConfig(use="nonexistent.module:FakeProvider"),
+        )
+        with pytest.raises(ValueError, match="nonexistent.module"):
+            resolve_authorization_provider(config)
+
+    def test_nonexistent_attribute_raises_with_path(self):
+        config = AuthorizationConfig(
+            enabled=True,
+            provider=AuthorizationProviderConfig(use="deerflow.authz.rbac:NonexistentProvider"),
+        )
+        with pytest.raises(ValueError, match="NonexistentProvider"):
+            resolve_authorization_provider(config)
+
+
+class TestProtocolConformance:
+    """Instance not satisfying AuthorizationProvider must fail."""
+
+    def test_non_protocol_class_raises(self):
+        """A class that constructs successfully but doesn't implement all
+        Protocol methods must fail the isinstance check."""
+        # `builtins:dict` constructs as empty dict(), which is not an
+        # AuthorizationProvider — it lacks name/authorize/aauthorize/filter_resources.
+        config = AuthorizationConfig(
+            enabled=True,
+            provider=AuthorizationProviderConfig(
+                use="builtins:dict",
+                config={},
+            ),
+        )
+        with pytest.raises(ValueError, match="AuthorizationProvider Protocol"):
+            resolve_authorization_provider(config)
+
+    def test_non_class_target_raises(self):
+        """A class path pointing to a non-class (e.g. a function) must fail
+        because resolve_variable is called with expected_type=type."""
+        config = AuthorizationConfig(
+            enabled=True,
+            provider=AuthorizationProviderConfig(
+                use="builtins:print",
+                config={},
+            ),
+        )
+        with pytest.raises(ValueError, match="Failed to resolve"):
+            resolve_authorization_provider(config)
+
+
+class TestRbacErrorPropagation:
+    """Factory must surface RBAC construction errors with class path and __cause__."""
+
+    def test_unknown_provider_config_key_surfaces_through_factory(self):
+        config = AuthorizationConfig(
+            enabled=True,
+            provider=AuthorizationProviderConfig(
+                use="deerflow.authz.rbac:RbacAuthorizationProvider",
+                config={"roles": {"user": {}}, "bogus": True},
+            ),
+        )
+        with pytest.raises(ValueError, match="RbacAuthorizationProvider.*bogus") as exc_info:
+            resolve_authorization_provider(config)
+        assert isinstance(exc_info.value.__cause__, ValueError)
+
+    def test_invalid_rbac_config_surfaces_class_path(self):
+        """RBAC construction failure (e.g. bad roles) must produce a ValueError
+        containing the class path."""
+        config = AuthorizationConfig(
+            enabled=True,
+            provider=AuthorizationProviderConfig(
+                use="deerflow.authz.rbac:RbacAuthorizationProvider",
+                config={"roles": "not a dict"},
+            ),
+        )
+        with pytest.raises(ValueError, match="RbacAuthorizationProvider"):
+            resolve_authorization_provider(config)
+
+    def test_invalid_rbac_config_preserves_cause(self):
+        """The original construction error must be chained as __cause__."""
+        config = AuthorizationConfig(
+            enabled=True,
+            provider=AuthorizationProviderConfig(
+                use="deerflow.authz.rbac:RbacAuthorizationProvider",
+                config={"roles": {"user": {"tools": {"allow": 42}}}},
+            ),
+        )
+        try:
+            resolve_authorization_provider(config)
+        except ValueError as err:
+            assert err.__cause__ is not None
+            assert isinstance(err.__cause__, ValueError)
+        else:
+            pytest.fail("Expected ValueError for invalid RBAC config")
+
+
+class TestNoFactoryInjection:
+    """Factory must not inject fail_closed or default_role into provider kwargs."""
+
+    def test_factory_does_not_inject_framework_params(self):
+        """The provider constructor should only receive config.config kwargs,
+        not fail_closed/default_role from AuthorizationConfig."""
+        config = AuthorizationConfig(
+            enabled=True,
+            fail_closed=False,
+            default_role="guest",
+            provider=AuthorizationProviderConfig(
+                use="deerflow.authz.rbac:RbacAuthorizationProvider",
+                config={"roles": {"guest": {"tools": {"allow": "*"}}}},
+            ),
+        )
+        # Should construct successfully without injecting fail_closed/default_role
+        provider = resolve_authorization_provider(config)
+        assert provider is not None
+
+
+class TestNoCaching:
+    """Factory must not cache instances."""
+
+    def test_each_call_returns_new_instance(self):
+        config = AuthorizationConfig(
+            enabled=True,
+            provider=AuthorizationProviderConfig(
+                use="deerflow.authz.rbac:RbacAuthorizationProvider",
+                config={"roles": {"admin": {"tools": {"allow": "*"}}}},
+            ),
+        )
+        p1 = resolve_authorization_provider(config)
+        p2 = resolve_authorization_provider(config)
+        assert p1 is not p2
+
+
+class TestDisabledNoImport:
+    """disabled must not trigger any import of the provider module."""
+
+    def test_disabled_does_not_import_invalid_path(self):
+        """Even with a garbage class path, disabled must return None
+        without raising ImportError."""
+        config = AuthorizationConfig(
+            enabled=False,
+            provider=AuthorizationProviderConfig(use="garbage.that.does.not.exist:Nope"),
+        )
+        # Must not raise
+        assert resolve_authorization_provider(config) is None
diff --git a/backend/tests/test_rbac_authorization_provider.py b/backend/tests/test_rbac_authorization_provider.py
new file mode 100644
index 000000000..e793d45d2
--- /dev/null
+++ b/backend/tests/test_rbac_authorization_provider.py
@@ -0,0 +1,416 @@
+"""Tests for the built-in RbacAuthorizationProvider."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from deerflow.authz.provider import AuthzRequest, Principal
+from deerflow.authz.rbac import RbacAuthorizationProvider
+
+# --- Helpers ---
+
+
+def _make_request(
+    *,
+    role: str = "user",
+    resource: str = "tool",
+    action: str = "call",
+    target: str = "bash",
+) -> AuthzRequest:
+    return AuthzRequest(
+        principal=Principal(role=role),
+        resource=resource,
+        action=action,
+        target=target,
+    )
+
+
+def _provider(roles: dict) -> RbacAuthorizationProvider:
+    return RbacAuthorizationProvider(roles=roles)
+
+
+# --- Allow semantics ---
+
+
+class TestAllowSemantics:
+    """Verify all forms of `allow` configuration."""
+
+    def test_wildcard_allow(self):
+        p = _provider({"user": {"tools": {"allow": "*"}}})
+        assert p.authorize(_make_request(target="bash")).allow is True
+        assert p.authorize(_make_request(target="write_file")).allow is True
+
+    def test_boolean_true_allow(self):
+        p = _provider({"user": {"tools": {"allow": True}}})
+        assert p.authorize(_make_request(target="bash")).allow is True
+
+    def test_boolean_false_deny_all(self):
+        p = _provider({"user": {"tools": {"allow": False}}})
+        assert p.authorize(_make_request(target="bash")).allow is False
+        assert p.authorize(_make_request(target="web_search")).allow is False
+
+    def test_list_allow(self):
+        p = _provider({"user": {"tools": {"allow": ["web_search", "read_file"]}}})
+        assert p.authorize(_make_request(target="web_search")).allow is True
+        assert p.authorize(_make_request(target="read_file")).allow is True
+        assert p.authorize(_make_request(target="bash")).allow is False
+
+    def test_empty_list_deny_all(self):
+        p = _provider({"user": {"tools": {"allow": []}}})
+        assert p.authorize(_make_request(target="bash")).allow is False
+
+    def test_allow_missing_defaults_to_allow_all(self):
+        """Missing `allow` means unrestricted (deny still applies)."""
+        p = _provider({"user": {"tools": {"deny": ["bash"]}}})
+        assert p.authorize(_make_request(target="bash")).allow is False
+        assert p.authorize(_make_request(target="web_search")).allow is True
+
+
+# --- Deny semantics ---
+
+
+class TestDenySemantics:
+    """Deny always wins over allow."""
+
+    def test_deny_overrides_wildcard(self):
+        p = _provider({"user": {"tools": {"allow": "*", "deny": ["bash"]}}})
+        assert p.authorize(_make_request(target="bash")).allow is False
+        assert p.authorize(_make_request(target="web_search")).allow is True
+
+    def test_deny_overrides_list_allow(self):
+        p = _provider({"user": {"tools": {"allow": ["bash", "web_search"], "deny": ["bash"]}}})
+        assert p.authorize(_make_request(target="bash")).allow is False
+        assert p.authorize(_make_request(target="web_search")).allow is True
+
+    def test_deny_overrides_boolean_true(self):
+        p = _provider({"user": {"tools": {"allow": True, "deny": ["bash"]}}})
+        assert p.authorize(_make_request(target="bash")).allow is False
+
+
+# --- Resource mapping ---
+
+
+class TestResourceMapping:
+    """tool → tools, model → models, etc."""
+
+    @pytest.mark.parametrize(
+        ("request_alias", "config_key"),
+        [
+            ("tool", "tools"),
+            ("model", "models"),
+            ("skill", "skills"),
+            ("mcp_server", "mcp_servers"),
+            ("route", "routes"),
+        ],
+    )
+    def test_reserved_request_alias_is_rejected(self, request_alias, config_key):
+        with pytest.raises(ValueError, match=rf"resource key '{request_alias}'.*use '{config_key}'"):
+            _provider({"user": {request_alias: {"allow": []}}})
+
+    def test_alias_is_rejected_when_mapped_key_is_also_configured(self):
+        with pytest.raises(ValueError, match=r"resource key 'tool'.*use 'tools'"):
+            _provider(
+                {
+                    "user": {
+                        "tools": {"allow": "*"},
+                        "tool": {"allow": []},
+                    }
+                }
+            )
+
+    def test_same_name_mapping_is_valid(self):
+        p = _provider({"user": {"sandbox": {"allow": []}}})
+        assert p.authorize(_make_request(resource="sandbox", target="default")).allow is False
+
+    def test_tool_maps_to_tools(self):
+        p = _provider({"user": {"tools": {"allow": ["web_search"]}}})
+        assert p.authorize(_make_request(resource="tool", target="web_search")).allow is True
+        assert p.authorize(_make_request(resource="tool", target="bash")).allow is False
+
+    def test_model_maps_to_models(self):
+        p = _provider({"user": {"models": {"allow": ["gpt-4o"]}}})
+        assert p.authorize(_make_request(resource="model", target="gpt-4o")).allow is True
+
+    def test_unknown_resource_uses_original_name(self):
+        p = _provider({"user": {"custom_resource": {"allow": ["item1"]}}})
+        assert p.authorize(_make_request(resource="custom_resource", target="item1")).allow is True
+
+    def test_resource_config_missing_means_unrestricted(self):
+        """If a role has no policy for a resource type, it's unrestricted."""
+        p = _provider({"user": {"tools": {"allow": ["web_search"]}}})
+        # No model policy configured → unrestricted
+        assert p.authorize(_make_request(resource="model", target="any")).allow is True
+
+
+# --- Role resolution ---
+
+
+class TestRoleResolution:
+    """Unknown and missing roles must fail."""
+
+    def test_known_role_works(self):
+        p = _provider({"admin": {"tools": {"allow": "*"}}, "user": {"tools": {"allow": []}}})
+        assert p.authorize(_make_request(role="admin", target="bash")).allow is True
+        assert p.authorize(_make_request(role="user", target="bash")).allow is False
+
+    def test_unknown_role_raises(self):
+        """Unknown role must raise ValueError, not return allow."""
+        p = _provider({"admin": {"tools": {"allow": "*"}}})
+        with pytest.raises(ValueError, match="Unknown role"):
+            p.authorize(_make_request(role="editor", target="bash"))
+
+    def test_missing_role_raises(self):
+        """None role must raise ValueError."""
+        p = _provider({"admin": {"tools": {"allow": "*"}}})
+        with pytest.raises(ValueError, match="no role"):
+            p.authorize(_make_request(role=None, target="bash"))
+
+    def test_empty_string_role_raises(self):
+        p = _provider({"admin": {"tools": {"allow": "*"}}})
+        with pytest.raises(ValueError, match="no role"):
+            p.authorize(_make_request(role="", target="bash"))
+
+
+# --- filter_resources ---
+
+
+class TestFilterResources:
+    """Batch visibility filter."""
+
+    def test_filter_preserves_order(self):
+        p = _provider({"user": {"tools": {"allow": ["web_search", "bash", "read_file"]}}})
+        result = p.filter_resources(Principal(role="user"), "tool", ["bash", "web_search", "write_file", "read_file"])
+        assert result == ["bash", "web_search", "read_file"]
+
+    def test_filter_no_duplicates_added(self):
+        p = _provider({"user": {"tools": {"allow": "*"}}})
+        result = p.filter_resources(Principal(role="user"), "tool", ["a", "b"])
+        assert result == ["a", "b"]
+
+    def test_filter_preserves_input_duplicates(self):
+        p = _provider({"user": {"tools": {"allow": "*"}}})
+        result = p.filter_resources(Principal(role="user"), "tool", ["a", "a", "b"])
+        assert result == ["a", "a", "b"]
+
+    def test_filter_does_not_modify_input(self):
+        p = _provider({"user": {"tools": {"allow": ["a"]}}})
+        candidates = ["a", "b", "c"]
+        p.filter_resources(Principal(role="user"), "tool", candidates)
+        assert candidates == ["a", "b", "c"]
+
+    def test_filter_unrestricted_when_no_policy(self):
+        p = _provider({"admin": {}})
+        result = p.filter_resources(Principal(role="admin"), "tool", ["a", "b"])
+        assert result == ["a", "b"]
+
+    def test_filter_consistent_with_authorize(self):
+        """filter_resources result must match per-item authorize decisions."""
+        p = _provider({"user": {"tools": {"allow": "*", "deny": ["bash"]}}})
+        candidates = ["bash", "web_search", "read_file", "write_file"]
+        filtered = p.filter_resources(Principal(role="user"), "tool", candidates)
+        per_item = [c for c in candidates if p.authorize(_make_request(target=c)).allow]
+        assert filtered == per_item
+
+    @pytest.mark.parametrize("role", [None, ""])
+    def test_filter_missing_role_raises(self, role):
+        """Visibility filtering propagates missing-role errors like authorize."""
+        p = _provider({"user": {"tools": {"allow": "*"}}})
+        with pytest.raises(ValueError, match="no role"):
+            p.filter_resources(Principal(role=role), "tool", ["bash"])
+
+    def test_filter_unknown_role_raises(self):
+        """Visibility filtering propagates unknown-role errors like authorize."""
+        p = _provider({"user": {"tools": {"allow": "*"}}})
+        with pytest.raises(ValueError, match="Unknown role"):
+            p.filter_resources(Principal(role="editor"), "tool", ["bash"])
+
+    @pytest.mark.parametrize("resource_type", [None, ""])
+    def test_filter_invalid_resource_type_raises(self, resource_type):
+        p = _provider({"user": {}})
+        with pytest.raises(ValueError, match="resource_type must be a non-empty string"):
+            p.filter_resources(Principal(role="user"), resource_type, ["bash"])
+
+    @pytest.mark.parametrize(
+        "roles",
+        [
+            {"user": {}},
+            {"user": {"tools": {"allow": "*"}}},
+            {"user": {"tools": {"allow": ["bash"]}}},
+        ],
+        ids=["unrestricted", "wildcard", "allow-list"],
+    )
+    @pytest.mark.parametrize("candidates", [["bash", None], ["bash", ""]], ids=["null", "empty"])
+    def test_filter_invalid_candidate_raises_for_every_policy_shape(self, roles, candidates):
+        p = _provider(roles)
+        with pytest.raises(ValueError, match=r"candidates\[1\] must be a non-empty string"):
+            p.filter_resources(Principal(role="user"), "tool", candidates)
+
+    @pytest.mark.parametrize("candidates", [None, ("bash",), "bash"])
+    def test_filter_non_list_candidates_raises(self, candidates):
+        p = _provider({"user": {"tools": {"allow": "*"}}})
+        with pytest.raises(ValueError, match="candidates must be a list"):
+            p.filter_resources(Principal(role="user"), "tool", candidates)
+
+
+# --- Request validation ---
+
+
+class TestRequestValidation:
+    """Malformed request identifiers must fail before any allow decision."""
+
+    @pytest.mark.parametrize(
+        "roles",
+        [
+            {"user": {}},
+            {"user": {"tools": {"allow": "*"}}},
+            {"user": {"tools": {"allow": ["bash"]}}},
+        ],
+        ids=["unrestricted", "wildcard", "allow-list"],
+    )
+    @pytest.mark.parametrize("target", [None, ""], ids=["null", "empty"])
+    def test_authorize_invalid_target_raises_for_every_policy_shape(self, roles, target):
+        p = _provider(roles)
+        with pytest.raises(ValueError, match="target must be a non-empty string"):
+            p.authorize(_make_request(target=target))
+
+    @pytest.mark.parametrize("resource", [None, ""], ids=["null", "empty"])
+    def test_authorize_invalid_resource_raises(self, resource):
+        p = _provider({"user": {"tools": {"allow": "*"}}})
+        with pytest.raises(ValueError, match="resource must be a non-empty string"):
+            p.authorize(_make_request(resource=resource))
+
+    @pytest.mark.parametrize("target", [None, ""], ids=["null", "empty"])
+    def test_aauthorize_invalid_target_matches_sync_validation(self, target):
+        p = _provider({"user": {"tools": {"allow": "*"}}})
+        with pytest.raises(ValueError, match="target must be a non-empty string"):
+            asyncio.run(p.aauthorize(_make_request(target=target)))
+
+
+# --- Sync / async parity ---
+
+
+class TestSyncAsyncParity:
+    def test_aauthorize_matches_authorize(self):
+        p = _provider({"user": {"tools": {"allow": "*", "deny": ["bash"]}}})
+        req = _make_request(target="bash")
+        sync = p.authorize(req)
+        async_ = asyncio.run(p.aauthorize(req))
+        assert sync.allow == async_.allow
+        assert sync.reasons[0].code == async_.reasons[0].code
+
+
+# --- Construction validation ---
+
+
+class TestConstructionValidation:
+    """Invalid config must fail at construction, not at request time."""
+
+    def test_unknown_provider_config_key_raises(self):
+        with pytest.raises(ValueError, match="unknown provider config keys.*bogus"):
+            RbacAuthorizationProvider(roles={"user": {}}, bogus=True)
+
+    def test_misspelled_roles_key_raises(self):
+        with pytest.raises(ValueError, match="unknown provider config keys.*rolez"):
+            RbacAuthorizationProvider(rolez={"user": {}})
+
+    def test_non_dict_roles_raises(self):
+        with pytest.raises(ValueError, match="roles must be a dict"):
+            RbacAuthorizationProvider(roles=["not", "a", "dict"])
+
+    def test_non_dict_role_config_raises(self):
+        with pytest.raises(ValueError, match="config must be a dict"):
+            RbacAuthorizationProvider(roles={"user": "not a dict"})
+
+    def test_non_dict_resource_policy_raises(self):
+        with pytest.raises(ValueError, match="must be a dict"):
+            RbacAuthorizationProvider(roles={"user": {"tools": "not a dict"}})
+
+    def test_invalid_allow_type_raises(self):
+        with pytest.raises(ValueError, match="allow must be"):
+            RbacAuthorizationProvider(roles={"user": {"tools": {"allow": 42}}})
+
+    def test_invalid_allow_string_raises(self):
+        with pytest.raises(ValueError, match="allow string must be"):
+            RbacAuthorizationProvider(roles={"user": {"tools": {"allow": "not_wildcard"}}})
+
+    def test_non_string_in_allow_list_raises(self):
+        with pytest.raises(ValueError, match="non-string"):
+            RbacAuthorizationProvider(roles={"user": {"tools": {"allow": ["ok", 42]}}})
+
+    def test_non_string_in_deny_list_raises(self):
+        with pytest.raises(ValueError, match="non-string"):
+            RbacAuthorizationProvider(roles={"user": {"tools": {"deny": ["ok", None]}}})
+
+    def test_empty_string_in_allow_list_raises(self):
+        with pytest.raises(ValueError, match="non-string or empty"):
+            RbacAuthorizationProvider(roles={"user": {"tools": {"allow": ["ok", ""]}}})
+
+    def test_invalid_deny_type_raises(self):
+        with pytest.raises(ValueError, match="deny must be"):
+            RbacAuthorizationProvider(roles={"user": {"tools": {"deny": 42}}})
+
+    def test_empty_role_name_raises(self):
+        with pytest.raises(ValueError, match="non-empty string"):
+            RbacAuthorizationProvider(roles={"": {"tools": {"allow": "*"}}})
+
+    def test_empty_resource_key_raises(self):
+        with pytest.raises(ValueError, match="invalid resource key"):
+            RbacAuthorizationProvider(roles={"user": {"": {"allow": "*"}}})
+
+    def test_explicit_null_allow_raises(self):
+        """`allow: null` must NOT be treated as missing — it's a config error."""
+        with pytest.raises(ValueError, match="allow must not be null"):
+            RbacAuthorizationProvider(roles={"user": {"tools": {"allow": None}}})
+
+    def test_explicit_null_deny_raises(self):
+        """`deny: null` must NOT be treated as missing — it's a config error."""
+        with pytest.raises(ValueError, match="deny must not be null"):
+            RbacAuthorizationProvider(roles={"user": {"tools": {"allow": "*", "deny": None}}})
+
+    def test_unknown_policy_key_raises(self):
+        """Misspelled keys (e.g. 'alow') must be rejected, not silently ignored."""
+        with pytest.raises(ValueError, match="unknown policy keys"):
+            RbacAuthorizationProvider(roles={"user": {"tools": {"alow": ["web_search"]}}})
+
+    def test_unknown_policy_key_with_valid_keys_raises(self):
+        """Unknown key alongside valid keys must still be rejected."""
+        with pytest.raises(ValueError, match="unknown policy keys"):
+            RbacAuthorizationProvider(roles={"user": {"tools": {"allow": "*", "permt": ["extra"]}}})
+
+    def test_mixed_type_unknown_keys_raises_value_error(self):
+        """Mixed-type unknown keys (e.g. str + int) must raise ValueError,
+        not TypeError from sorted() comparison failure."""
+        with pytest.raises(ValueError, match="unknown policy keys"):
+            RbacAuthorizationProvider(roles={"user": {"tools": {1: "bad", "other": "bad"}}})
+
+
+# --- Config immutability ---
+
+
+class TestConfigImmutability:
+    """Provider must not be affected by post-construction config mutation."""
+
+    def test_mutating_config_after_construction_does_not_change_behavior(self):
+        roles_config = {"user": {"tools": {"allow": ["bash"]}}}
+        p = RbacAuthorizationProvider(roles=roles_config)
+        # Mutate the original config
+        roles_config["user"]["tools"]["allow"] = ["web_search"]
+        roles_config["admin"] = {"tools": {"allow": "*"}}
+        # Provider should still use the original compiled policy
+        assert p.authorize(_make_request(target="bash")).allow is True
+        assert p.authorize(_make_request(target="web_search")).allow is False
+        # Unknown role added to config should not be known to provider
+        with pytest.raises(ValueError, match="Unknown role"):
+            p.authorize(_make_request(role="admin", target="bash"))
+
+
+# --- Protocol conformance ---
+
+
+class TestProtocolConformance:
+    def test_rbac_is_authorization_provider(self):
+        from deerflow.authz.provider import AuthorizationProvider
+
+        assert isinstance(RbacAuthorizationProvider(roles={}), AuthorizationProvider)
diff --git a/docs/plans/2026-07-10-pluggable-authorization-implementation-notes.md b/docs/plans/2026-07-10-pluggable-authorization-implementation-notes.md
index 5b6f61484..11b1208eb 100644
--- a/docs/plans/2026-07-10-pluggable-authorization-implementation-notes.md
+++ b/docs/plans/2026-07-10-pluggable-authorization-implementation-notes.md
@@ -200,6 +200,54 @@ Phase 1 最低验证要求：
 - **延期：** RBAC provider、provider factory、Layer 1 过滤、Layer 2 自动接线移至
   Phase 1A-2 / Phase 1B。
 
+### 2026-07-17 — Phase 1A-2 / 内置 RBAC provider 与 provider factory
+
+- **背景：** Phase 1A-1 建立了可信 Principal 链路，但没有策略引擎。
+  Phase 1A-2 实现内置 RBAC provider 和统一 provider factory。
+- **决策：** `RbacAuthorizationProvider` 在构造时完成全部配置校验并编译为
+  不可变结构（`frozenset` / sentinel `_ALL`）。请求路径只做 O(1) membership 检查。
+- **决策：** deny 永远优先于 allow，无论 allow 是 `"*"`、`True`、列表还是缺失。
+- **决策：** 未知角色和缺失角色抛 `ValueError`（不返回 allow），由执行层
+  根据 `fail_closed` 决定。
+- **决策：** 资源名使用显式映射（`tool → tools`，`model → models` 等），
+  不通过加 `s` 猜测。配置中的保留请求别名（如 `tool`）在构造期拒绝，并提示使用
+  对应配置键（如 `tools`），防止策略被存储在永远无法命中的键下。未知 resource
+  使用原名查找；未配置时视为"不受限"。
+- **决策：** `resolve_authorization_provider()` 是唯一 provider 解析入口。
+  disabled 时返回 `None`（不 import provider 模块）；enabled 但缺少 provider
+  时抛 `ValueError`。不缓存实例。不注入 `fail_closed` 或 `default_role`。
+- **决策：** 内置和自定义 provider 使用完全相同的 `resolve_variable` class-path
+  解析路径，无特殊分支。
+- **证据：** 66 tests passed（51 RBAC + 15 factory，其中 5 条为 malformed-policy
+  回归测试）。
+- **兼容性：** 无运行时行为变化（`authorization.enabled: false`）。不修改
+  `config.example.yaml`，不 bump `config_version`。
+- **延期：** Layer 1 工具过滤、Layer 2 自动接线、DeerFlowClient、RBAC 配置示例
+  移至 Phase 1B。
+- **Phase 1B 注意：** 已知角色缺少某个 resource policy 时语义是“不受限”，不是
+  fail-closed。配置示例必须明确提醒，并枚举部署方希望限制的每种 resource。
+- **Phase 1B 注意：** 内置 RBAC 当前按 role + resource + target 决策，不区分
+  `AuthzRequest.action`；`policy_id` 也是稳定但粗粒度的
+  `rbac:allow` / `rbac:deny` / `rbac:unrestricted`。接入审计日志前应决定是否通过
+  更具体的 policy id 或 decision metadata 记录 role / resource / target。
+
+### 2026-07-20 — Phase 1A-2 / PR #4260 请求边界收口
+
+- **背景：** review 发现 `request.target` 未经运行时校验；通配符策略会允许
+  `None` 或空字符串，而列表策略会拒绝，形成依赖策略形态的不一致结果。进一步审查
+  发现无效 resource 和批量过滤候选项也存在相同的“不受限/通配符路径放行”风险。
+- **决策：** 内置 RBAC 在请求边界要求 resource、resource type、target 和每个
+  candidate 都是非空字符串；`filter_resources()` 还要求 candidates 是 list。
+  非法输入统一抛 `ValueError`，不能进入 `rbac:unrestricted` 或通配符 allow 路径。
+- **决策：** `filter_resources()` 对缺失/未知角色继续与 `authorize()` 一致地抛
+  `ValueError`，不在 provider 内静默返回空列表。Phase 1B 集成层负责按
+  `fail_closed` 处理 provider 异常，避免隐藏身份或部署配置错误。
+- **证据：** 90 tests passed（75 RBAC + 15 factory），覆盖 unrestricted、
+  wildcard、allow-list、同步/异步、非法 resource/target/candidates，以及
+  `filter_resources()` 的缺失/未知角色错误语义。
+- **兼容性：** 只拒绝不符合 `AuthzRequest` / `filter_resources` 类型契约的运行时
+  输入；Phase 1A-2 仍未接入运行时，`authorization.enabled: false` 行为不变。
+
 ### 新记录模板
 
 ```markdown