Runtime — Aquilia Documentation
Comprehensive guide and documentation for Runtime in the Aquilia framework. View API reference, examples, and implementation patterns.
WebSockets / Runtime WebSocket Runtime The AquilaSockets runtime manages connection lifecycles, upgrades ASGI HTTP connections to WebSockets, decodes incoming messages, runs auth guards, and coordinates pub/sub scaling. Handshake & Lifespan Cycle When a client connects to a SocketController route, the runtime coordinates the lifecycle through these phases: , , , ].map((item, i) => ( ))} Built-in Security Guards Guards inherit from the base SocketGuard protocol class and are applied via @Guard decorators. They can intercept either the initial HTTP handshake or individual incoming client messages. 1. HandshakeAuthGuard Authenticates and authorizes connections during the initial HTTP upgrade handshake phase. If the check fails, the connection is aborted immediately. from aquilia.sockets import Guard, HandshakeAuthGuard # Require a valid user identity that is flagged as an admin @Guard(HandshakeAuthGuard( require_identity=True, require_session=True, allowed_identity_types=["admin"] )) class AdminSocketController(SocketController): pass 2. OriginGuard Validates the upgrade request's origin header against a list of allowed endpoints to prevent Cross-Site WebSocket Hijacking (CSWSH) attacks. from aquilia.sockets import Guard, OriginGuard # Reject any requests originating from unrecognized domains @Guard(OriginGuard(allowed_origins=["https://myapp.com", "https://*.myapp.com"])) class SecureController(SocketController): pass 3. MessageAuthGuard Periodically re-authenticates the user token during the lifespan of the WebSocket connection, ensuring revoked tokens disconnect active users within the set interval. from aquilia.sockets import Guard, MessageAuthGuard # Re-authenticate the active token every 5 minutes (300 seconds) @Guard(MessageAuthGuard(check_interval=300)) class LongLivedController(SocketController): pass 4. RateLimitGuard Applies rate limit controls on a per-connection basis to prevent clients from flooding handlers with excessive message frequencies. from aquilia.sockets import Guard, RateLimitGuard, Event class MessageRateController(SocketController): # Limit message.send events to a maximum of 5 payloads per second @Event("message.send") @Guard(RateLimitGuard(messages_per_second=5)) async def on_send(self, conn, payload): pass Writing Custom Socket Guards To write a custom guard, subclass SocketGuard and implement check_handshake (ran once at upgrade) or check_message (ran on every incoming message event): from aquilia.sockets import SocketGuard, Connection, ConnectionScope, MessageEnvelope from aquilia.faults import WS_FORBIDDEN # Socket fault status code class RoomAccessGuard(SocketGuard): def __init__(self, role: str): self.role = role async def check_handshake(self, scope: ConnectionScope) -> None: # Check permissions early from request headers or path parameters room_id = scope.path_params.get("room_id") user = scope.identity if not user or not user.has_room_role(room_id, self.role): # Abort the upgrade phase immediately raise WS_FORBIDDEN("You do not have access to this room.") async def check_message(self, conn: Connection, envelope: MessageEnvelope) -> None: # Check permissions on specific incoming messages if envelope.event == "room.admin_action" and not conn.identity.is_staff: raise WS_FORBIDDEN("Staff only action.") Socket Controllers Adapters )
Go to Homepage