5. Auth & Policy
Protect operations declaratively. The bus enforces it — same check for REST, MCP, CLI, and direct calls.
Two Layers of Protection
| Layer | What it does | Declared on |
|---|---|---|
| Auth enforcement | Checks identity before a runnable executes | @runnable (per operation) |
| Invoke policy | Controls which bus targets a component can call | kernel.set_policy() (per component) |
Auth asks “is this caller allowed?” Policy asks “is this component allowed to call that target?”
requires_action
@runnable("create", params=CreateParams, description="Create document",
requires_action="docs.write")
async def create_doc(self, params):
return {"id": "new-doc"}Before this runnable executes, the bus extracts __auth_token__ from params, authenticates via IAuth, and checks authorization for "docs.write". Denied → PermissionError.
IAuth?
IAuth is a kernel contract (defined in signalpy.kernel.contracts) — it’s how the bus asks “who is this caller, and may they perform action X?” You provide the answer by writing an AuthProvider component that @provides(IAuth). The kernel ships an example JWT-based one; in dev, you can omit it entirely (see “Without IAuth” at the end).
# Without token → PermissionError
await kernel.bus.invoke("doc-svc.create", {"title": "New"})
# With token → works if authorized
await kernel.bus.invoke("doc-svc.create", {
"title": "New",
"__auth_token__": "my-token",
})requires_role
@runnable("delete", params=DeleteParams, description="Delete document",
requires_role="admin")
async def delete_doc(self, params):
return {"deleted": True}Checks that the caller’s identity has the "admin" role.
Auth is enforced at the bus level, not the transport level. Same check whether invoked via REST, MCP, CLI, or another component’s self.rt.invoke().
Invoke Policy
Restrict what bus targets a component can call:
kernel.set_policy("untrusted-app", {
"invoke_allow": ["public.*"],
"invoke_deny": ["admin.*"],
"audit": True,
})Enforced on self.rt.invoke(). Direct kernel.bus.invoke() bypasses policy.
self.rt.invoke(target, params) is the policy-enforced path that components use to call each other. The kernel knows the caller (the component owning self.rt), so it can check the policy table.
kernel.bus.invoke(target, params) is the lower-level path: same dispatch, but no caller identity. Use it from tests, scripts, or transport adapters where the policy doesn’t apply (the transport itself is what enforces auth at the edge).
Auth (requires_action / requires_role) runs on both paths because it depends on the __auth_token__ in params, not the caller. Policy runs on self.rt.invoke() only.
Without IAuth
If no AuthProvider is in the kernel, requires_action and requires_role are silently skipped. Develop without auth, add it later without changing component code.