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

authorization: use SQL tables to store user info 

Add a UserInfo class which provides type annotations for each member.
This class easily converts to a dict or tuple, simplifying conversion
for usage in SQL statements.

Signed-off-by:  Eric Callahan <arksine.code@gmail.com>
This commit is contained in:
Eric Callahan 2024-05-10 12:45:54 -04:00
parent 3f0d20ed8c
commit eddf47e4a3
2 changed files with 187 additions and 118 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
moonraker/common.py
View File

@ -10,8 +10,9 @@ import logging
import copy import copy
import re import re
import inspect import inspect
import dataclasses
import time
from enum import Enum, Flag, auto from enum import Enum, Flag, auto
from dataclasses import dataclass
from abc import ABCMeta, abstractmethod from abc import ABCMeta, abstractmethod
from .utils import ServerError, Sentinel from .utils import ServerError, Sentinel
from .utils import json_wrapper as jsonw from .utils import json_wrapper as jsonw
@ -177,7 +178,24 @@ class RenderableTemplate(metaclass=ABCMeta):
async def render_async(self, context: Dict[str, Any] = {}) -> str: async def render_async(self, context: Dict[str, Any] = {}) -> str:
... ...
@dataclass(frozen=True) @dataclasses.dataclass
class UserInfo:
username: str
password: str
created_on: float = dataclasses.field(default_factory=time.time)
salt: str = ""
source: str = "moonraker"
jwt_secret: Optional[str] = None
jwk_id: Optional[str] = None
groups: List[str] = dataclasses.field(default_factory=lambda: ["admin"])
def as_tuple(self) -> Tuple[Any, ...]:
return dataclasses.astuple(self)
def as_dict(self) -> Dict[str, Any]:
return dataclasses.asdict(self)
@dataclasses.dataclass(frozen=True)
class APIDefinition: class APIDefinition:
endpoint: str endpoint: str
http_path: str http_path: str

279
moonraker/components/authorization.py
View File

@ -20,7 +20,7 @@ import logging
from tornado.web import HTTPError from tornado.web import HTTPError
from libnacl.sign import Signer, Verifier from libnacl.sign import Signer, Verifier
from ..utils import json_wrapper as jsonw from ..utils import json_wrapper as jsonw
from ..common import RequestType, TransportType from ..common import RequestType, TransportType, SqlTableDefinition, UserInfo
# Annotation imports # Annotation imports
from typing import ( from typing import (
@ -39,6 +39,7 @@ if TYPE_CHECKING:
from .websockets import WebsocketManager from .websockets import WebsocketManager
from tornado.httputil import HTTPServerRequest from tornado.httputil import HTTPServerRequest
from .database import MoonrakerDatabase as DBComp from .database import MoonrakerDatabase as DBComp
from .database import DBProviderWrapper
from .ldap import MoonrakerLDAP from .ldap import MoonrakerLDAP
IPAddr = Union[ipaddress.IPv4Address, ipaddress.IPv6Address] IPAddr = Union[ipaddress.IPv4Address, ipaddress.IPv6Address]
IPNetwork = Union[ipaddress.IPv4Network, ipaddress.IPv6Network] IPNetwork = Union[ipaddress.IPv4Network, ipaddress.IPv6Network]
@ -60,6 +61,7 @@ TRUSTED_CONNECTION_TIMEOUT = 3600
FQDN_CACHE_TIMEOUT = 84000 FQDN_CACHE_TIMEOUT = 84000
PRUNE_CHECK_TIME = 300. PRUNE_CHECK_TIME = 300.
USER_TABLE = "authorized_users"
AUTH_SOURCES = ["moonraker", "ldap"] AUTH_SOURCES = ["moonraker", "ldap"]
HASH_ITER = 100000 HASH_ITER = 100000
API_USER = "_API_KEY_USER_" API_USER = "_API_KEY_USER_"
@ -71,6 +73,47 @@ JWT_HEADER = {
'typ': "JWT" 'typ': "JWT"
} }
class UserSqlDefinition(SqlTableDefinition):
name = USER_TABLE
prototype = (
f"""
{USER_TABLE} (
username TEXT PRIMARY KEY NOT NULL,
password TEXT NOT NULL,
created_on REAL NOT NULL,
salt TEXT NOT NULL,
source TEXT NOT NULL,
jwt_secret TEXT,
jwk_id TEXT,
groups pyjson
)
"""
)
version = 1
def migrate(self, last_version: int, db_provider: DBProviderWrapper) -> None:
if last_version == 0:
users: Dict[str, Dict[str, Any]]
users = db_provider.get_namespace("authorized_users")
api_user = users.pop(API_USER, {})
user_vals: List[Tuple[Any, ...]] = [
UserInfo(
username=API_USER,
password=api_user.get("api_key", uuid.uuid4().hex),
created_on=api_user.get("created_on", time.time())
).as_tuple()
]
for user in users.values():
user_vals.append(UserInfo(**user).as_tuple())
placeholders = ",".join("?" * len(user_vals[0]))
conn = db_provider.connection
with conn:
conn.executemany(
f"INSERT OR IGNORE INTO {USER_TABLE} VALUES({placeholders})",
user_vals
)
db_provider.wipe_local_namespace("authorized_users")
class Authorization: class Authorization:
def __init__(self, config: ConfigHelper) -> None: def __init__(self, config: ConfigHelper) -> None:
self.server = config.get_server() self.server = config.get_server()
@ -97,62 +140,13 @@ class Authorization:
" however [ldap] section failed to load or not configured" " however [ldap] section failed to load or not configured"
) )
database: DBComp = self.server.lookup_component('database') database: DBComp = self.server.lookup_component('database')
database.register_local_namespace('authorized_users', forbidden=True) self.user_table = database.register_table(UserSqlDefinition())
self.user_db = database.wrap_namespace('authorized_users') self.users: Dict[str, UserInfo] = {}
self.users: Dict[str, Dict[str, Any]] = self.user_db.as_dict()
api_user: Optional[Dict[str, Any]] = self.users.get(API_USER, None)
if api_user is None:
self.api_key = uuid.uuid4().hex self.api_key = uuid.uuid4().hex
self.users[API_USER] = {
'username': API_USER,
'api_key': self.api_key,
'created_on': time.time()
}
else:
self.api_key = api_user['api_key']
hi = self.server.get_host_info() hi = self.server.get_host_info()
self.issuer = f"http://{hi['hostname']}:{hi['port']}" self.issuer = f"http://{hi['hostname']}:{hi['port']}"
self.public_jwks: Dict[str, Dict[str, Any]] = {} self.public_jwks: Dict[str, Dict[str, Any]] = {}
for username, user_info in list(self.users.items()): self.trusted_users: Dict[IPAddr, Dict[str, Any]] = {}
if username == API_USER:
# Validate the API User
for item in ["username", "api_key", "created_on"]:
if item not in user_info:
self.users[API_USER] = {
'username': API_USER,
'api_key': self.api_key,
'created_on': time.time()
}
break
continue
else:
# validate created users
valid = True
for item in ["username", "password", "salt", "created_on"]:
if item not in user_info:
logging.info(
f"Authorization: User {username} does not "
f"contain field {item}, removing")
del self.users[username]
valid = False
break
if not valid:
continue
# generate jwks for valid users
if 'jwt_secret' in user_info:
try:
priv_key = self._load_private_key(user_info['jwt_secret'])
jwk_id = user_info['jwk_id']
except (self.server.error, KeyError):
logging.info("Invalid key found for user, removing")
user_info.pop('jwt_secret', None)
user_info.pop('jwk_id', None)
self.users[username] = user_info
continue
self.public_jwks[jwk_id] = self._generate_public_jwk(priv_key)
# sync user changes to the database
self.user_db.sync(self.users)
self.trusted_users: Dict[IPAddr, Any] = {}
self.oneshot_tokens: Dict[str, OneshotToken] = {} self.oneshot_tokens: Dict[str, OneshotToken] = {}
# Get allowed cors domains # Get allowed cors domains
@ -276,17 +270,68 @@ class Authorization:
"authorization:user_logged_out", event_type="logout" "authorization:user_logged_out", event_type="logout"
) )
def _sync_user(self, username: str) -> None:
self.user_db[username] = self.users[username]
async def component_init(self) -> None: async def component_init(self) -> None:
# Populate users from database
cursor = await self.user_table.execute(f"SELECT * FROM {USER_TABLE}")
self.users = {row[0]: UserInfo(**dict(row)) for row in await cursor.fetchall()}
need_sync = self._initialize_users()
if need_sync:
await self._sync_user_table()
self.prune_timer.start(delay=PRUNE_CHECK_TIME) self.prune_timer.start(delay=PRUNE_CHECK_TIME)
async def _sync_user(self, username: str) -> None:
user = self.users[username]
vals = user.as_tuple()
placeholders = ",".join("?" * len(vals))
async with self.user_table as tx:
await tx.execute(
f"REPLACE INTO {USER_TABLE} VALUES({placeholders})", vals
)
async def _sync_user_table(self) -> None:
async with self.user_table as tx:
await tx.execute(f"DELETE FROM {USER_TABLE}")
user_vals: List[Tuple[Any, ...]]
user_vals = [user.as_tuple() for user in self.users.values()]
if not user_vals:
return
placeholders = ",".join("?" * len(user_vals[0]))
await tx.executemany(
f"INSERT INTO {USER_TABLE} VALUES({placeholders})", user_vals
)
def _initialize_users(self) -> bool:
need_sync = False
api_user: Optional[UserInfo] = self.users.get(API_USER, None)
if api_user is None:
need_sync = True
self.users[API_USER] = UserInfo(username=API_USER, password=self.api_key)
else:
self.api_key = api_user.password
for username, user_info in list(self.users.items()):
if username == API_USER:
continue
# generate jwks for valid users
if user_info.jwt_secret is not None:
try:
priv_key = self._load_private_key(user_info.jwt_secret)
jwk_id = user_info.jwk_id
assert jwk_id is not None
except (self.server.error, KeyError, AssertionError):
logging.info("Invalid jwk found for user, removing")
user_info.jwt_secret = None
user_info.jwk_id = None
self.users[username] = user_info
need_sync = True
continue
self.public_jwks[jwk_id] = self._generate_public_jwk(priv_key)
return need_sync
async def _handle_apikey_request(self, web_request: WebRequest) -> str: async def _handle_apikey_request(self, web_request: WebRequest) -> str:
if web_request.get_request_type() == RequestType.POST: if web_request.get_request_type() == RequestType.POST:
self.api_key = uuid.uuid4().hex self.api_key = uuid.uuid4().hex
self.users[API_USER]['api_key'] = self.api_key self.users[API_USER].password = self.api_key
self._sync_user(API_USER) await self._sync_user(API_USER)
return self.api_key return self.api_key
async def _handle_oneshot_request(self, web_request: WebRequest) -> str: async def _handle_oneshot_request(self, web_request: WebRequest) -> str:
@ -322,10 +367,12 @@ class Authorization:
if username in RESERVED_USERS: if username in RESERVED_USERS:
raise self.server.error( raise self.server.error(
f"Invalid log out request for user {username}") f"Invalid log out request for user {username}")
self.users[username].pop("jwt_secret", None) jwk_id: Optional[str] = self.users[username].jwk_id
jwk_id: str = self.users[username].pop("jwk_id", None) self.users[username].jwt_secret = None
self._sync_user(username) self.users[username].jwk_id = None
if jwk_id is not None:
self.public_jwks.pop(jwk_id, None) self.public_jwks.pop(jwk_id, None)
await self._sync_user(username)
eventloop = self.server.get_event_loop() eventloop = self.server.get_event_loop()
eventloop.delay_callback( eventloop.delay_callback(
.005, self.server.send_event, "authorization:user_logged_out", .005, self.server.send_event, "authorization:user_logged_out",
@ -363,16 +410,16 @@ class Authorization:
user_info = self.decode_jwt(refresh_token, token_type="refresh") user_info = self.decode_jwt(refresh_token, token_type="refresh")
except Exception: except Exception:
raise self.server.error("Invalid Refresh Token", 401) raise self.server.error("Invalid Refresh Token", 401)
username: str = user_info['username'] username: str = user_info.username
if 'jwt_secret' not in user_info or "jwk_id" not in user_info: if user_info.jwt_secret is None or user_info.jwk_id is None:
raise self.server.error("User not logged in", 401) raise self.server.error("User not logged in", 401)
private_key = self._load_private_key(user_info['jwt_secret']) private_key = self._load_private_key(user_info.jwt_secret)
jwk_id: str = user_info['jwk_id'] jwk_id: str = user_info.jwk_id
token = self._generate_jwt(username, jwk_id, private_key) token = self._generate_jwt(username, jwk_id, private_key)
return { return {
'username': username, 'username': username,
'token': token, 'token': token,
'source': user_info.get("source", "moonraker"), 'source': user_info.source,
'action': 'user_jwt_refresh' 'action': 'user_jwt_refresh'
} }
@ -399,7 +446,7 @@ class Authorization:
return await self._login_jwt_user(web_request, create=True) return await self._login_jwt_user(web_request, create=True)
elif req_type == RequestType.DELETE: elif req_type == RequestType.DELETE:
# Delete User # Delete User
return self._delete_jwt_user(web_request) return await self._delete_jwt_user(web_request)
raise self.server.error("Invalid Request Method") raise self.server.error("Invalid Request Method")
async def _handle_list_request(self, async def _handle_list_request(self,
@ -407,12 +454,12 @@ class Authorization:
) -> Dict[str, List[Dict[str, Any]]]: ) -> Dict[str, List[Dict[str, Any]]]:
user_list = [] user_list = []
for user in self.users.values(): for user in self.users.values():
if user['username'] == API_USER: if user.username == API_USER:
continue continue
user_list.append({ user_list.append({
'username': user['username'], 'username': user.username,
'source': user.get("source", "moonraker"), 'source': user.source,
'created_on': user['created_on'] 'created_on': user.created_on
}) })
return { return {
'users': user_list 'users': user_list
@ -440,8 +487,8 @@ class Authorization:
raise self.server.error("Invalid Password") raise self.server.error("Invalid Password")
new_hashed_pass = hashlib.pbkdf2_hmac( new_hashed_pass = hashlib.pbkdf2_hmac(
'sha256', new_pass.encode(), salt, HASH_ITER).hex() 'sha256', new_pass.encode(), salt, HASH_ITER).hex()
self.users[username]['password'] = new_hashed_pass self.users[username].password = new_hashed_pass
self._sync_user(username) await self._sync_user(username)
return { return {
'username': username, 'username': username,
'action': "user_password_reset" 'action': "user_password_reset"
@ -457,7 +504,7 @@ class Authorization:
).lower() ).lower()
if source not in AUTH_SOURCES: if source not in AUTH_SOURCES:
raise self.server.error(f"Invalid 'source': {source}") raise self.server.error(f"Invalid 'source': {source}")
user_info: Dict[str, Any] user_info: UserInfo
if username in RESERVED_USERS: if username in RESERVED_USERS:
raise self.server.error( raise self.server.error(
f"Invalid Request for user {username}") f"Invalid Request for user {username}")
@ -477,15 +524,14 @@ class Authorization:
salt = secrets.token_bytes(32) salt = secrets.token_bytes(32)
hashed_pass = hashlib.pbkdf2_hmac( hashed_pass = hashlib.pbkdf2_hmac(
'sha256', password.encode(), salt, HASH_ITER).hex() 'sha256', password.encode(), salt, HASH_ITER).hex()
user_info = { user_info = UserInfo(
'username': username, username=username,
'password': hashed_pass, password=hashed_pass,
'salt': salt.hex(), salt=salt.hex(),
'source': source, source=source,
'created_on': time.time() )
}
self.users[username] = user_info self.users[username] = user_info
self._sync_user(username) await self._sync_user(username)
action = "user_created" action = "user_created"
if source == "ldap": if source == "ldap":
# Dont notify user created # Dont notify user created
@ -495,30 +541,32 @@ class Authorization:
if username not in self.users: if username not in self.users:
raise self.server.error(f"Unregistered User: {username}") raise self.server.error(f"Unregistered User: {username}")
user_info = self.users[username] user_info = self.users[username]
auth_src = user_info.get("source", "moonraker") auth_src = user_info.source
if auth_src != source: if auth_src != source:
raise self.server.error( raise self.server.error(
f"Moonraker cannot authenticate user '{username}', must " f"Moonraker cannot authenticate user '{username}', must "
f"specify source '{auth_src}'", 401 f"specify source '{auth_src}'", 401
) )
salt = bytes.fromhex(user_info['salt']) salt = bytes.fromhex(user_info.salt)
hashed_pass = hashlib.pbkdf2_hmac( hashed_pass = hashlib.pbkdf2_hmac(
'sha256', password.encode(), salt, HASH_ITER).hex() 'sha256', password.encode(), salt, HASH_ITER).hex()
action = "user_logged_in" action = "user_logged_in"
if hashed_pass != user_info['password']: if hashed_pass != user_info.password:
raise self.server.error("Invalid Password") raise self.server.error("Invalid Password")
jwt_secret_hex: Optional[str] = user_info.get('jwt_secret', None) jwt_secret_hex: Optional[str] = user_info.jwt_secret
if jwt_secret_hex is None: if jwt_secret_hex is None:
private_key = Signer() private_key = Signer()
jwk_id = base64url_encode(secrets.token_bytes()).decode() jwk_id = base64url_encode(secrets.token_bytes()).decode()
user_info['jwt_secret'] = private_key.hex_seed().decode() user_info.jwt_secret = private_key.hex_seed().decode()
user_info['jwk_id'] = jwk_id user_info.jwk_id = jwk_id
self.users[username] = user_info self.users[username] = user_info
self._sync_user(username) await self._sync_user(username)
self.public_jwks[jwk_id] = self._generate_public_jwk(private_key) self.public_jwks[jwk_id] = self._generate_public_jwk(private_key)
else: else:
private_key = self._load_private_key(jwt_secret_hex) private_key = self._load_private_key(jwt_secret_hex)
jwk_id = user_info['jwk_id'] if user_info.jwk_id is None:
user_info.jwk_id = base64url_encode(secrets.token_bytes()).decode()
jwk_id = user_info.jwk_id
token = self._generate_jwt(username, jwk_id, private_key) token = self._generate_jwt(username, jwk_id, private_key)
refresh_token = self._generate_jwt( refresh_token = self._generate_jwt(
username, jwk_id, private_key, token_type="refresh", username, jwk_id, private_key, token_type="refresh",
@ -531,16 +579,16 @@ class Authorization:
"authorization:user_created", "authorization:user_created",
{'username': username}) {'username': username})
elif conn is not None: elif conn is not None:
conn.user_info = user_info conn.user_info = user_info.as_dict()
return { return {
'username': username, 'username': username,
'token': token, 'token': token,
'source': user_info.get("source", "moonraker"), 'source': user_info.source,
'refresh_token': refresh_token, 'refresh_token': refresh_token,
'action': action 'action': action
} }
def _delete_jwt_user(self, web_request: WebRequest) -> Dict[str, str]: async def _delete_jwt_user(self, web_request: WebRequest) -> Dict[str, str]:
username: str = web_request.get_str('username') username: str = web_request.get_str('username')
current_user = web_request.get_current_user() current_user = web_request.get_current_user()
if current_user is not None: if current_user is not None:
@ -551,13 +599,16 @@ class Authorization:
if username in RESERVED_USERS: if username in RESERVED_USERS:
raise self.server.error( raise self.server.error(
f"Invalid Request for reserved user {username}") f"Invalid Request for reserved user {username}")
user_info: Optional[Dict[str, Any]] = self.users.get(username) user_info: Optional[UserInfo] = self.users.get(username)
if user_info is None: if user_info is None:
raise self.server.error(f"No registered user: {username}") raise self.server.error(f"No registered user: {username}")
if 'jwk_id' in user_info: if user_info.jwk_id is not None:
self.public_jwks.pop(user_info['jwk_id'], None) self.public_jwks.pop(user_info.jwk_id, None)
del self.users[username] del self.users[username]
del self.user_db[username] async with self.user_table as tx:
await tx.execute(
f"DELETE FROM {USER_TABLE} WHERE username = ?", (username,)
)
event_loop = self.server.get_event_loop() event_loop = self.server.get_event_loop()
event_loop.delay_callback( event_loop.delay_callback(
.005, self.server.send_event, .005, self.server.send_event,
@ -595,7 +646,7 @@ class Authorization:
def decode_jwt( def decode_jwt(
self, token: str, token_type: str = "access", check_exp: bool = True self, token: str, token_type: str = "access", check_exp: bool = True
) -> Dict[str, Any]: ) -> UserInfo:
message, sig = token.rsplit('.', maxsplit=1) message, sig = token.rsplit('.', maxsplit=1)
enc_header, enc_payload = message.split('.') enc_header, enc_payload = message.split('.')
header: Dict[str, Any] = jsonw.loads(base64url_decode(enc_header)) header: Dict[str, Any] = jsonw.loads(base64url_decode(enc_header))
@ -626,7 +677,7 @@ class Authorization:
raise self.server.error("JWT Expired", 401) raise self.server.error("JWT Expired", 401)
# get user # get user
user_info: Optional[Dict[str, Any]] = self.users.get( user_info: Optional[UserInfo] = self.users.get(
payload.get('username', ""), None) payload.get('username', ""), None)
if user_info is None: if user_info is None:
raise self.server.error("Unknown user", 401) raise self.server.error("Unknown user", 401)
@ -641,13 +692,13 @@ class Authorization:
raise self.server.error( raise self.server.error(
f"Failed to decode JWT: {e}", 401 f"Failed to decode JWT: {e}", 401
) from e ) from e
return user_info return user_info.as_dict()
def validate_api_key(self, api_key: str) -> Dict[str, Any]: def validate_api_key(self, api_key: str) -> Dict[str, Any]:
if not self.enable_api_key: if not self.enable_api_key:
raise self.server.error("API Key authentication is disabled", 401) raise self.server.error("API Key authentication is disabled", 401)
if api_key and api_key == self.api_key: if api_key and api_key == self.api_key:
return self.users[API_USER] return self.users[API_USER].as_dict()
raise self.server.error("Invalid API Key", 401) raise self.server.error("Invalid API Key", 401)
def _load_private_key(self, secret: str) -> Signer: def _load_private_key(self, secret: str) -> Signer:
@ -709,7 +760,7 @@ class Authorization:
def _check_json_web_token( def _check_json_web_token(
self, request: HTTPServerRequest, required: bool = True self, request: HTTPServerRequest, required: bool = True
) -> Optional[Dict[str, Any]]: ) -> Optional[UserInfo]:
auth_token: Optional[str] = request.headers.get("Authorization") auth_token: Optional[str] = request.headers.get("Authorization")
if auth_token is None: if auth_token is None:
auth_token = request.headers.get("X-Access-Token") auth_token = request.headers.get("X-Access-Token")
@ -757,23 +808,21 @@ class Authorization:
async def _check_trusted_connection( async def _check_trusted_connection(
self, ip: Optional[IPAddr] self, ip: Optional[IPAddr]
) -> Optional[Dict[str, Any]]: ) -> Optional[UserInfo]:
if ip is not None: if ip is not None:
curtime = time.time() curtime = time.time()
exp_time = curtime + TRUSTED_CONNECTION_TIMEOUT exp_time = curtime + TRUSTED_CONNECTION_TIMEOUT
if ip in self.trusted_users: if ip in self.trusted_users:
self.trusted_users[ip]['expires_at'] = exp_time self.trusted_users[ip]["expires_at"] = exp_time
return self.trusted_users[ip] return self.trusted_users[ip]["user"]
elif await self._check_authorized_ip(ip): elif await self._check_authorized_ip(ip):
logging.info( logging.info(
f"Trusted Connection Detected, IP: {ip}") f"Trusted Connection Detected, IP: {ip}")
self.trusted_users[ip] = { self.trusted_users[ip] = {
'username': TRUSTED_USER, "user": UserInfo(TRUSTED_USER, "", curtime),
'password': None, "expires_at": exp_time
'created_on': curtime,
'expires_at': exp_time
} }
return self.trusted_users[ip] return self.trusted_users[ip]["user"]
return None return None
def _check_oneshot_token(self, def _check_oneshot_token(self,
@ -805,7 +854,7 @@ class Authorization:
# Check JSON Web Token # Check JSON Web Token
jwt_user = self._check_json_web_token(request, auth_required) jwt_user = self._check_json_web_token(request, auth_required)
if jwt_user is not None: if jwt_user is not None:
return jwt_user return jwt_user.as_dict()
try: try:
ip = ipaddress.ip_address(request.remote_ip) # type: ignore ip = ipaddress.ip_address(request.remote_ip) # type: ignore
@ -825,7 +874,7 @@ class Authorization:
if self.enable_api_key: if self.enable_api_key:
key: Optional[str] = request.headers.get("X-Api-Key") key: Optional[str] = request.headers.get("X-Api-Key")
if key and key == self.api_key: if key and key == self.api_key:
return self.users[API_USER] return self.users[API_USER].as_dict()
# If the force_logins option is enabled and at least one user is created # If the force_logins option is enabled and at least one user is created
# then trusted user authentication is disabled # then trusted user authentication is disabled
@ -837,8 +886,10 @@ class Authorization:
# Check if IP is trusted. If this endpoint doesn't require authentication # Check if IP is trusted. If this endpoint doesn't require authentication
# then it is acceptable to return None # then it is acceptable to return None
trusted_user = await self._check_trusted_connection(ip) trusted_user = await self._check_trusted_connection(ip)
if trusted_user is not None or not auth_required: if trusted_user is not None:
return trusted_user return trusted_user.as_dict()
if not auth_required:
return None
raise HTTPError(401, "Unauthorized") raise HTTPError(401, "Unauthorized")