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 are designed to intercept the initial HTTP handshake. Deprecation Notice SocketGuard.check_message is deprecated. MessageAuthGuard and RateLimitGuard (per-message guards) have never executed. Use the new Socket Middleware subsystem instead for per-message processing. check_handshake remains the supported way to gate a connection. 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 Routing & Fault Behaviors Parameterized Route Matching WebSocket controllers fully support parameterized routes (e.g., @Socket("/chat/:room")). The path parameters are automatically extracted and available within the connection scope via conn.scope.path_params. Policy Violation Close Codes When security requirements are not met (such as WS_AUTH_REQUIRED, WS_FORBIDDEN, or WS_ORIGIN_NOT_ALLOWED), the socket terminates automatically using WebSocket close code 1008 (Policy Violation), accurately reflecting unauthorized access per RFC 6455. Writing Custom Socket Guards To write a custom guard, subclass SocketGuard and implement check_handshake (ran once at upgrade): from aquilia.sockets import SocketGuard, ConnectionScope from aquilia.faults import WS_FORBIDDEN # Closes socket with code 1008 (Policy Violation) 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.") Socket Controllers Adapters )
Go to Homepage