3diy
/
moonraker
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

websockets: add support for API Key authentication 

Signed-off-by:  Eric Callahan <arksine.code@gmail.com>
This commit is contained in:
Eric Callahan 2023-01-18 16:44:58 -05:00
parent eca4c7e438
commit 06ec5541e3
No known key found for this signature in database
GPG Key ID: 5A1EB336DFB4C71B
2 changed files with 19 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
moonraker/components/authorization.py
View File

@ -615,6 +615,13 @@ class Authorization:
) from e ) from e
return user_info return user_info
def validate_api_key(self, api_key: str) -> Dict[str, Any]:
if not self.enable_api_key:
raise self.server.error("API Key authentication is disabled", 401)
if api_key and api_key == self.api_key:
return self.users[API_USER]
raise self.server.error("Invalid API Key", 401)
def _load_private_key(self, secret: str) -> Signer: def _load_private_key(self, secret: str) -> Signer:
try: try:
key = Signer(bytes.fromhex(secret)) key = Signer(bytes.fromhex(secret))

16
moonraker/websockets.py
View File

@ -376,7 +376,10 @@ class WebsocketManager(APITransport):
async def _handle_identify(self, args: Dict[str, Any]) -> Dict[str, int]: async def _handle_identify(self, args: Dict[str, Any]) -> Dict[str, int]:
sc: BaseSocketClient = args["_socket_"] sc: BaseSocketClient = args["_socket_"]
sc.authenticate(token=args.get("access_token", None)) sc.authenticate(
token=args.get("access_token", None),
api_key=args.get("api_key", None)
)
if sc.identified: if sc.identified:
raise self.server.error( raise self.server.error(
f"Connection already identified: {sc.client_data}" f"Connection already identified: {sc.client_data}"
@ -569,15 +572,20 @@ class BaseSocketClient(Subscribable):
self.queue_busy = True self.queue_busy = True
self.eventloop.register_callback(self._write_messages) self.eventloop.register_callback(self._write_messages)
def authenticate(self, path: str = "", token: Optional[str] = None) -> None: def authenticate(
self, path: str = "",
token: Optional[str] = None,
api_key: Optional[str] = None
) -> None:
if not self._need_auth: if not self._need_auth:
return return
auth: AuthComp = self.server.lookup_component("authorization", None) auth: AuthComp = self.server.lookup_component("authorization", None)
if auth is None: if auth is None:
return return
if token is not None: if token is not None:
user_info = auth.validate_jwt(token) self.user_info = auth.validate_jwt(token)
self.user_info = user_info elif api_key is not None:
self.user_info = auth.validate_api_key(api_key)
elif not auth.is_path_permitted(path): elif not auth.is_path_permitted(path):
raise self.server.error("Unauthorized", 401) raise self.server.error("Unauthorized", 401)