Skip to content

BaseHubArbiter

autogen.beta.network.hub.arbiter.BaseHubArbiter #

No-op base implementation. Override only the gates you care about.

Mirrors :class:BaseHubListener: every method returns :class:Allow so subclasses opting in to a single gate (e.g. authorize_register for tenant-quota checks) don't have to implement the rest of the surface.

For the rule-based behavior the hub had inline before this seam existed, use :class:RuleBasedArbiter (the default) and compose / chain via subclassing.

authorize_send `async` #

authorize_send(envelope, sender, sender_rule, recipients)

Source code in autogen/beta/network/hub/arbiter.py

async def authorize_send(
    self,
    envelope: "Envelope",  # noqa: ARG002
    sender: "Passport",  # noqa: ARG002
    sender_rule: "Rule",  # noqa: ARG002
    recipients: list["Passport"],  # noqa: ARG002
) -> Decision:
    return Allow()

authorize_inbox `async` #

authorize_inbox(envelope, recipient, recipient_rule, current_pending)

Source code in autogen/beta/network/hub/arbiter.py

async def authorize_inbox(
    self,
    envelope: "Envelope",  # noqa: ARG002
    recipient: "Passport",  # noqa: ARG002
    recipient_rule: "Rule",  # noqa: ARG002
    current_pending: int,  # noqa: ARG002
) -> Decision:
    return Allow()

authorize_dispatch `async` #

authorize_dispatch(envelope, sender, recipient, recipient_rule)

Source code in autogen/beta/network/hub/arbiter.py

async def authorize_dispatch(
    self,
    envelope: "Envelope",  # noqa: ARG002
    sender: "Passport",  # noqa: ARG002
    recipient: "Passport",  # noqa: ARG002
    recipient_rule: "Rule",  # noqa: ARG002
) -> Decision:
    return Allow()

authorize_channel_open `async` #

authorize_channel_open(manifest, creator, creator_rule, invitees, invitee_rules, active_creator_channels)

Source code in autogen/beta/network/hub/arbiter.py

async def authorize_channel_open(
    self,
    manifest: "ChannelManifest",  # noqa: ARG002
    creator: "Passport",  # noqa: ARG002
    creator_rule: "Rule",  # noqa: ARG002
    invitees: list["Passport"],  # noqa: ARG002
    invitee_rules: list["Rule"],  # noqa: ARG002
    active_creator_channels: int,  # noqa: ARG002
) -> Decision:
    return Allow()

authorize_register `async` #

authorize_register(passport, resume, rule)

Source code in autogen/beta/network/hub/arbiter.py

async def authorize_register(
    self,
    passport: "Passport",  # noqa: ARG002
    resume: "Resume",  # noqa: ARG002
    rule: "Rule",  # noqa: ARG002
) -> Decision:
    return Allow()

resolve_unknown_audience `async` #

resolve_unknown_audience(envelope, unknown_ids)

Source code in autogen/beta/network/hub/arbiter.py

async def resolve_unknown_audience(
    self,
    envelope: "Envelope",  # noqa: ARG002
    unknown_ids: list[str],  # noqa: ARG002
) -> list[str] | None:
    return None