GithubFollow/One-KVM
1
0
Fork 0
mirror of https://github.com/mofeng-git/One-KVM.git synced 2025-12-17 03:40:24 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

improved auth logging

Browse Source
This commit is contained in:
Maxim Devaev 2025-02-12 12:51:15 +02:00
parent 83c352a900
commit c66c97afd4
1 changed files with 22 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
kvmd/apps/kvmd/auth.py
View File

@ -66,24 +66,26 @@ class AuthManager: # pylint: disable=too-many-instance-attributes
totp_secret_path: str, totp_secret_path: str,
) -> None: ) -> None:
logger = get_logger(0)
self.__enabled = enabled self.__enabled = enabled
if not enabled: if not enabled:
get_logger().warning("AUTHORIZATION IS DISABLED") logger.warning("AUTHORIZATION IS DISABLED")
assert expire >= 0 assert expire >= 0
self.__expire = expire self.__expire = expire
if expire > 0: if expire > 0:
get_logger().info("Maximum user session time is limited: %s", logger.info("Maximum user session time is limited: %s",
self.__format_seconds(expire)) self.__format_seconds(expire))
self.__unauth_paths = frozenset(unauth_paths) # To speed up self.__unauth_paths = frozenset(unauth_paths) # To speed up
for path in self.__unauth_paths: for path in self.__unauth_paths:
get_logger().warning("Authorization is disabled for API %r", path) logger.warning("Authorization is disabled for API %r", path)
self.__int_service: (BaseAuthService | None) = None self.__int_service: (BaseAuthService | None) = None
if enabled: if enabled:
self.__int_service = get_auth_service_class(int_type)(**int_kwargs) self.__int_service = get_auth_service_class(int_type)(**int_kwargs)
get_logger().info("Using internal auth service %r", logger.info("Using internal auth service %r",
self.__int_service.get_plugin_name()) self.__int_service.get_plugin_name())
self.__force_int_users = force_int_users self.__force_int_users = force_int_users
@ -91,7 +93,7 @@ class AuthManager: # pylint: disable=too-many-instance-attributes
self.__ext_service: (BaseAuthService | None) = None self.__ext_service: (BaseAuthService | None) = None
if enabled and ext_type: if enabled and ext_type:
self.__ext_service = get_auth_service_class(ext_type)(**ext_kwargs) self.__ext_service = get_auth_service_class(ext_type)(**ext_kwargs)
get_logger().info("Using external auth service %r", logger.info("Using external auth service %r",
self.__ext_service.get_plugin_name()) self.__ext_service.get_plugin_name())
self.__totp_secret_path = totp_secret_path self.__totp_secret_path = totp_secret_path
@ -113,6 +115,7 @@ class AuthManager: # pylint: disable=too-many-instance-attributes
assert user assert user
assert self.__enabled assert self.__enabled
assert self.__int_service assert self.__int_service
logger = get_logger(0)
if self.__totp_secret_path: if self.__totp_secret_path:
with open(self.__totp_secret_path) as file: with open(self.__totp_secret_path) as file:
@ -120,7 +123,7 @@ class AuthManager: # pylint: disable=too-many-instance-attributes
if secret: if secret:
code = passwd[-6:] code = passwd[-6:]
if not pyotp.TOTP(secret).verify(code, valid_window=1): if not pyotp.TOTP(secret).verify(code, valid_window=1):
get_logger().error("Got access denied for user %r by TOTP", user) logger.error("Got access denied for user %r by TOTP", user)
return False return False
passwd = passwd[:-6] passwd = passwd[:-6]
@ -132,9 +135,9 @@ class AuthManager: # pylint: disable=too-many-instance-attributes
pname = service.get_plugin_name() pname = service.get_plugin_name()
ok = (await service.authorize(user, passwd)) ok = (await service.authorize(user, passwd))
if ok: if ok:
get_logger().info("Authorized user %r via auth service %r", user, pname) logger.info("Authorized user %r via auth service %r", user, pname)
else: else:
get_logger().error("Got access denied for user %r from auth service %r", user, pname) logger.error("Got access denied for user %r from auth service %r", user, pname)
return ok return ok
async def login(self, user: str, passwd: str, expire: int) -> (str | None): async def login(self, user: str, passwd: str, expire: int) -> (str | None):
@ -150,7 +153,7 @@ class AuthManager: # pylint: disable=too-many-instance-attributes
expire_ts=self.__make_expire_ts(expire), expire_ts=self.__make_expire_ts(expire),
) )
self.__sessions[token] = session self.__sessions[token] = session
get_logger().info("Logged in user %r; expire=%s, sessions_now=%d", get_logger(0).info("Logged in user %r; expire=%s, sessions_now=%d",
session.user, session.user,
self.__format_expire_ts(session.expire_ts), self.__format_expire_ts(session.expire_ts),
self.__get_sessions_number(session.user)) self.__get_sessions_number(session.user))
@ -214,7 +217,7 @@ class AuthManager: # pylint: disable=too-many-instance-attributes
if session.user == user: if session.user == user:
count += 1 count += 1
del self.__sessions[key_t] del self.__sessions[key_t]
get_logger().info("Logged out user %r; sessions_closed=%d", user, count) get_logger(0).info("Logged out user %r; sessions_closed=%d", user, count)
def check(self, token: str) -> (str | None): def check(self, token: str) -> (str | None):
assert self.__enabled assert self.__enabled
@ -229,7 +232,7 @@ class AuthManager: # pylint: disable=too-many-instance-attributes
return session.user return session.user
else: else:
del self.__sessions[token] del self.__sessions[token]
get_logger().info("The session of user %r is expired; sessions_left=%d", get_logger(0).info("The session of user %r is expired; sessions_left=%d",
session.user, session.user,
self.__get_sessions_number(session.user)) self.__get_sessions_number(session.user))
return None return None