test/network/test_rule_enforcement.py
Rama main · réplica de lectura del repositorio original.
# Copyright (c) 2026, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors
#
# SPDX-License-Identifier: Apache-2.0
"""Rule enforcement — access blocks, concurrency caps, TTL parsing.
Validates every enforced field on ``Rule``:
* ``access.outbound_to`` / ``access.inbound_from`` — glob over peer name
* ``limits.delegation_depth`` — max envelope ``depth`` (0 disables)
* ``limits.max_concurrent_channels`` — capped on creator (0 disables)
* ``limits.max_concurrent_tasks`` — capped on owner via ``observe_task``
* ``limits.inbox.max_pending`` — backpressure on dispatch (substantive only)
* ``limits.channel_ttl_default`` — drives ``ChannelMetadata.expires_at``
The per-minute throttle in ``RateBlock`` is stored but not enforced
by the in-process hub (see ``RateBlock`` docstring), so it isn't
tested here.
"""
import asyncio
from datetime import datetime
import pytest
from ag2 import Agent
from ag2.knowledge import MemoryKnowledgeStore
from ag2.network import (
EV_TEXT,
AccessBlock,
AccessDeniedError,
Envelope,
Hub,
HubClient,
InboxFull,
LimitsBlock,
LocalLink,
Passport,
Resume,
Rule,
)
from ag2.network.rule import InboxBlock, parse_duration
from ag2.task import TaskMetadata, TaskSpec, TaskState
from ._helpers import ScriptedConfig
def _agent(name: str) -> Agent:
return Agent(name=name, config=ScriptedConfig())
# ── parse_duration ──────────────────────────────────────────────────────────
@pytest.mark.parametrize(
"value,expected",
[
("30s", 30),
("15m", 900),
("2h", 7200),
("4h", 14400),
("1d", 86400),
("0s", 0),
(0, 0),
(3600, 3600),
("", 0),
("60", 60),
("3600", 3600),
],
)
def test_parse_duration_valid(value, expected) -> None:
assert parse_duration(value) == expected
@pytest.mark.parametrize("value", ["abc", "5x", "10y", "fast"])
def test_parse_duration_invalid_raises(value) -> None:
with pytest.raises(ValueError):
parse_duration(value)
# ── Access: outbound_to + inbound_from ──────────────────────────────────────
@pytest.mark.asyncio
async def test_outbound_to_glob_allows_pattern() -> None:
"""outbound_to supports glob via fnmatch — `bot-*` matches `bot-bob`.
The sender's own access rule must NOT block protocol broadcasts
that include the sender in their audience (``EV_CHANNEL_OPENED`` /
``EV_CHANNEL_CLOSED``). The hub skips the outbound check when
``recipient_id == sender_id``, so a creator with restrictive
``outbound_to`` (not including themselves) can still create
channels and receive their own channel-state notifications.
"""
hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0, expectation_sweep_interval=0)
alice = await hub.register(
_agent("alice"),
rule=Rule(access=AccessBlock(outbound_to=["bot-*"])), # alice not in own whitelist
)
bob = await hub.register(_agent("bot-bob"))
eve = await hub.register(_agent("user-eve"))
# Reaching bot-bob is allowed.
channel_ok = await alice.open(type="conversation", target="bot-bob")
await channel_ok.send("greetings", audience=[bob.agent_id])
# Reaching user-eve is denied.
envelope = Envelope(
channel_id=channel_ok.channel_id,
sender_id=alice.agent_id,
audience=[eve.agent_id],
event_type=EV_TEXT,
event_data={"text": "x"},
)
with pytest.raises(AccessDeniedError):
await alice.send_envelope(envelope)
await hub.close()
@pytest.mark.asyncio
async def test_outbound_to_self_send_always_allowed() -> None:
"""A sender posting to their own agent_id passes the outbound check
even when their ``outbound_to`` whitelist excludes their own name."""
hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0, expectation_sweep_interval=0)
link = LocalLink(hub)
hc = HubClient(link, hub=hub)
alice = await hc.register(
_agent("alice"),
Passport(name="alice"),
Resume(),
rule=Rule(access=AccessBlock(outbound_to=["nobody-*"])),
)
bob = await hc.register(_agent("bob"), Passport(name="bob"), Resume())
# Alice can self-only-broadcast (audience contains alice + bob).
# Without the self-skip in the outbound check, this would raise
# AccessDeniedError before reaching bob.
# Use a channel manifest that allows arbitrary text — conversation works.
# First open with bob (bob is in audience for invite — but alice's
# outbound_to=["nobody-*"] doesn't match "bob" → invite denied).
# So instead, register bob under a name matching the whitelist:
await hc.unregister_agent(bob.agent_id)
await hc.register(_agent("nobody-bob"), Passport(name="nobody-bob"), Resume())
channel = await alice.open(type="conversation", target="nobody-bob")
# Audience includes alice — without the self-skip fix, opening would fail.
assert channel.metadata.state.value == "active"
await hc.close()
await hub.close()
@pytest.mark.asyncio
async def test_inbound_from_blocks_dispatch_not_post() -> None:
"""inbound_from is a recipient-side filter — the sender's post
succeeds (WAL appends) but no NotifyFrame reaches the blocked-from
recipient.
Setup uses ``set_rule`` mid-channel because pre-channel
``inbound_from`` blocks are now caught at ``create_channel`` time
by the fail-fast check. This test specifically validates the
in-flight dispatch filter.
"""
hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0, expectation_sweep_interval=0)
alice = await hub.register(_agent("alice"))
bob = await hub.register(_agent("bob"))
received_by_bob: list[Envelope] = []
# Capture wrapper — preserves the default handler behaviour
# (auto-ack invites, etc.) but lets us observe what arrived.
original_handler = bob._on_envelope
assert original_handler is not None
async def capture(env: Envelope) -> None:
received_by_bob.append(env)
await original_handler(env)
bob.on_envelope(capture)
channel = await alice.open(type="conversation", target="bob")
assert sum(1 for e in received_by_bob if e.event_type == EV_TEXT) == 0
# Tighten bob's inbound filter — alice no longer matches.
await bob.set_rule(Rule(access=AccessBlock(inbound_from=["nobody-*"])))
await channel.send("filtered text", audience=[bob.agent_id])
await asyncio.sleep(0.05)
# WAL has the text; bob's handler did NOT see it.
wal = await hub.read_wal(channel.channel_id)
assert any(e.event_type == EV_TEXT and e.event_data["text"] == "filtered text" for e in wal)
post_text_received = [e for e in received_by_bob if e.event_data.get("text") == "filtered text"]
assert post_text_received == []
await hub.close()
@pytest.mark.asyncio
async def test_inbound_from_blocks_create_channel_fast_fails() -> None:
"""Creating a channel with an invitee whose ``inbound_from`` blocks
the creator must raise ``AccessDeniedError`` immediately, not hang
until ``invite_ack_timeout``.
Without the pre-flight check in ``create_channel``, the dispatch
path silently filters the invite, the recipient never sees it,
never acks, and the creator times out with a generic ``ProtocolError``.
"""
hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0, expectation_sweep_interval=0)
alice = await hub.register(_agent("alice"))
await hub.register(
_agent("bob"),
rule=Rule(access=AccessBlock(inbound_from=["carol-*"])),
)
with pytest.raises(AccessDeniedError, match="does not accept inbound"):
await alice.open(type="conversation", target="bob")
# No channel leaked into the registry.
assert await hub.list_channels() == []
await hub.close()
# ── delegation_depth ────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_delegation_depth_at_cap_accepted_above_rejected() -> None:
"""depth == cap is accepted (`>` check); depth == cap+1 is rejected."""
hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0, expectation_sweep_interval=0)
alice = await hub.register(
_agent("alice"),
rule=Rule(limits=LimitsBlock(delegation_depth=3)),
)
bob = await hub.register(_agent("bob"))
channel = await alice.open(type="conversation", target="bob")
# depth=3 (== cap) → ok
await channel.send("at-cap", audience=[bob.agent_id], depth=3)
# depth=4 (> cap) → denied
with pytest.raises(AccessDeniedError, match="delegation_depth"):
await channel.send("over-cap", audience=[bob.agent_id], depth=4)
await hub.close()
@pytest.mark.asyncio
async def test_delegation_depth_zero_disables_cap() -> None:
"""delegation_depth=0 means unlimited."""
hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0, expectation_sweep_interval=0)
alice = await hub.register(
_agent("alice"),
rule=Rule(limits=LimitsBlock(delegation_depth=0)),
)
bob = await hub.register(_agent("bob"))
channel = await alice.open(type="conversation", target="bob")
# Even an absurd depth should pass.
await channel.send("any-depth", audience=[bob.agent_id], depth=99999)
await hub.close()
# ── max_concurrent_channels ─────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_max_concurrent_channels_cap_blocks_new_creates() -> None:
hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0, expectation_sweep_interval=0)
alice = await hub.register(
_agent("alice"),
rule=Rule(limits=LimitsBlock(max_concurrent_channels=2)),
)
await hub.register(_agent("bob"))
await hub.register(_agent("carol"))
await hub.register(_agent("dave"))
# 2 concurrent channels ok.
s1 = await alice.open(type="conversation", target="bob")
await alice.open(type="conversation", target="carol")
# Third attempt → AccessDeniedError before any persistence.
with pytest.raises(AccessDeniedError, match="max_concurrent_channels"):
await alice.open(type="conversation", target="dave")
# Closing one frees the slot.
await s1.close()
await alice.open(type="conversation", target="dave") # ok now
await hub.close()
@pytest.mark.asyncio
async def test_max_concurrent_channels_zero_disables() -> None:
hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0, expectation_sweep_interval=0)
alice = await hub.register(
_agent("alice"),
rule=Rule(limits=LimitsBlock(max_concurrent_channels=0)),
)
for name in ("bob", "carol", "dave", "erin"):
await hub.register(_agent(name))
# Open 4 — no cap.
for name in ("bob", "carol", "dave", "erin"):
await alice.open(type="conversation", target=name)
channels = await hub.list_channels(agent_id=alice.agent_id)
assert len(channels) == 4
await hub.close()
# ── max_concurrent_tasks (via observe_task) ─────────────────────────────────
@pytest.mark.asyncio
async def test_max_concurrent_tasks_blocks_observe() -> None:
hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0, expectation_sweep_interval=0)
alice = await hub.register(
_agent("alice"),
rule=Rule(limits=LimitsBlock(max_concurrent_tasks=2)),
)
# Two non-terminal tasks ok.
for i in range(2):
await hub.observe_task(
TaskMetadata(
task_id=f"task-{i}",
owner_id=alice.agent_id,
spec=TaskSpec(title=f"t{i}"),
state=TaskState.RUNNING,
)
)
# Third → denied.
with pytest.raises(AccessDeniedError, match="max_concurrent_tasks"):
await hub.observe_task(
TaskMetadata(
task_id="task-2",
owner_id=alice.agent_id,
spec=TaskSpec(title="t2"),
state=TaskState.RUNNING,
)
)
# Terminating one frees the slot.
await hub.update_task("task-0", state=TaskState.COMPLETED)
await hub.observe_task(
TaskMetadata(
task_id="task-2",
owner_id=alice.agent_id,
spec=TaskSpec(title="t2"),
state=TaskState.RUNNING,
)
)
await hub.close()
# ── inbox.max_pending ───────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_inbox_max_pending_rejects_when_full() -> None:
"""Substantive sends to a recipient at inbox capacity raise InboxFull.
The counter increments on dispatch; bob never replies (ScriptedConfig
returns ""), so it never decrements. Once at cap, alice's next send
is rejected before WAL append.
"""
hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0, expectation_sweep_interval=0)
alice = await hub.register(_agent("alice"))
bob = await hub.register(
_agent("bob"),
rule=Rule(limits=LimitsBlock(inbox=InboxBlock(max_pending=2))),
)
channel = await alice.open(type="conversation", target="bob")
await asyncio.sleep(0.05) # let channel-protocol dispatch settle
# Cap=2 means the counter must be < 2 to allow a new send.
# Protocol envelopes don't increment, so two substantive sends
# bring bob's count to 2 → third raises.
await channel.send("msg-1", audience=[bob.agent_id])
await channel.send("msg-2", audience=[bob.agent_id])
with pytest.raises(InboxFull):
await channel.send("msg-3", audience=[bob.agent_id])
await hub.close()
@pytest.mark.asyncio
async def test_inbox_protocol_events_bypass_capacity() -> None:
"""Protocol envelopes (invite/ack/open/close) must always reach the
recipient, regardless of inbox capacity. Otherwise a blocked invite
ack would deadlock the channel machine."""
hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0, expectation_sweep_interval=0)
alice = await hub.register(_agent("alice"))
# Cap of 1 — easy to overflow with substantive but invites must
# still be dispatched.
await hub.register(
_agent("bob"),
rule=Rule(limits=LimitsBlock(inbox=InboxBlock(max_pending=1))),
)
# If invites failed the inbox cap, this would hang at invite ack
# timeout. Success here confirms the protocol-event bypass.
channel = await alice.open(type="conversation", target="bob")
assert channel.metadata.state.value == "active"
await hub.close()
# ── channel_ttl_default ─────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_channel_ttl_default_drives_expires_at() -> None:
hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0, expectation_sweep_interval=0)
alice = await hub.register(
_agent("alice"),
rule=Rule(limits=LimitsBlock(channel_ttl_default="1h")),
)
await hub.register(_agent("bob"))
channel = await alice.open(type="conversation", target="bob")
assert channel.metadata.expires_at is not None
created = datetime.fromisoformat(channel.metadata.created_at)
expires = datetime.fromisoformat(channel.metadata.expires_at)
delta = (expires - created).total_seconds()
# 1h ± 1s
assert 3599 <= delta <= 3601
await hub.close()
@pytest.mark.asyncio
async def test_channel_ttl_per_channel_override_wins() -> None:
hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0, expectation_sweep_interval=0)
alice = await hub.register(
_agent("alice"),
rule=Rule(limits=LimitsBlock(channel_ttl_default="1h")),
)
await hub.register(_agent("bob"))
channel = await alice.open(type="conversation", target="bob", ttl="30m")
expires = datetime.fromisoformat(channel.metadata.expires_at)
created = datetime.fromisoformat(channel.metadata.created_at)
delta = (expires - created).total_seconds()
assert 1799 <= delta <= 1801
await hub.close()
@pytest.mark.asyncio
async def test_channel_ttl_zero_no_expiry() -> None:
"""ttl=0 → no expires_at stamped."""
hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0, expectation_sweep_interval=0)
alice = await hub.register(_agent("alice"))
await hub.register(_agent("bob"))
channel = await alice.open(type="conversation", target="bob", ttl=0)
assert channel.metadata.expires_at is None
await hub.close()