fix(messaging): close fail-open when agent_slug lookup returns None

I1: post_to_channel raised no error if get_agent_slug returned None
(unknown/deleted agent), and send_message's 'if agent_slug' skipped
the validate_channel_access call. Same fail-open class the prior
commit was fixing, just narrower window. Now post_to_channel raises
ChannelAccessDeniedError directly when the slug lookup fails — say()
already converts that to a clean not_authorized Envelope.

I2: 'writable channels for your role' wording was misleading because
get_agent_channels resolves by slug, not role. Replaced with
'channels you may write to' for clarity.
This commit is contained in:
Renn F
2026-05-03 05:46:02 +02:00
parent 7907988e52
commit 8b43ed98be
3 changed files with 47 additions and 6 deletions
+1 -1
View File
@@ -226,7 +226,7 @@ class ContentActions:
message=(
f"agent '{e.agent_id}' may not write to channel '{e.channel_slug}'"
),
remediate=f"writable channels for your role: {writable_str}",
remediate=f"channels you may write to: {writable_str}",
context_briefing={},
)
return Envelope.ok(
+13 -1
View File
@@ -1810,14 +1810,26 @@ class MessagingService(BaseService):
Channel access is enforced inside `send_message` (channel writers
list + role rules) when `agent_slug` is supplied — this adapter
looks up the slug from `agent_id` and forwards it so the check
always runs.
always runs. If the slug lookup returns None (unknown / removed
agent), we fail closed by raising ChannelAccessDeniedError rather
than letting send_message silently skip validate_channel_access.
"""
from roboco.enforcement.channel_access import ChannelAccessDeniedError
from roboco.services.repositories import get_agent_slug
channel = await self.get_channel_by_slug_or_raise(channel_slug)
group = await self._default_group_for_channel(channel)
session = await self.get_or_create_active_session(cast("UUID", group.id))
agent_slug = await get_agent_slug(self.session, agent_id)
if agent_slug is None:
raise ChannelAccessDeniedError(
agent_id=str(agent_id),
channel_slug=channel_slug,
action="write",
message=(
f"agent {agent_id} not found; cannot validate channel access"
),
)
return await self.send_message(
MessageCreateRequest(
agent_id=agent_id,
+33 -4
View File
@@ -66,11 +66,13 @@ async def test_say_returns_not_authorized_envelope_on_access_denied() -> None:
assert body["error"] == "not_authorized"
assert "announcements" in body["message"]
# Remediation should hint at the agent's writable channels (or that
# they have none). The role/list is resolved from CHANNEL_ACCESS via
# get_agent_channels using the agent's slug.
# Remediation should hint at the channels the agent may write to (or
# that they have none). The list is resolved from CHANNEL_ACCESS via
# get_agent_channels using the agent's slug — note the wording is
# slug-keyed ("channels you may write to") rather than role-keyed,
# since CHANNEL_ACCESS is keyed by slug not role.
assert body["remediate"] is not None
assert "writable" in body["remediate"].lower()
assert "channels you may write to" in body["remediate"].lower()
@pytest.mark.asyncio
@@ -90,3 +92,30 @@ async def test_say_success_path_returns_posted_envelope() -> None:
assert body["status"] == "posted"
assert body["next"] == "continue"
msg_svc.post_to_channel.assert_awaited_once()
@pytest.mark.asyncio
async def test_say_returns_not_authorized_when_agent_lookup_fails() -> None:
"""If get_agent_slug returns None (deleted agent), say must fail closed.
Pins the I1 fix: post_to_channel raises ChannelAccessDeniedError directly
when the slug lookup fails, so send_message's `if agent_slug:` can no
longer skip validate_channel_access for unknown/removed agents. The
Envelope conversion in say() carries that through to the agent.
"""
aid = uuid4()
msg_svc = AsyncMock()
msg_svc.post_to_channel.side_effect = ChannelAccessDeniedError(
agent_id=str(aid),
channel_slug="dev-all",
action="write",
message="agent not found",
)
deps = _make_deps(messaging=msg_svc)
actions = ContentActions(deps)
env = await actions.say(agent_id=aid, channel="dev-all", text="hi")
body = env.as_dict()
assert body["error"] == "not_authorized"
assert "dev-all" in body["message"]