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

kvmd/auth/disabled -> kvmd/auth/enabled

Browse Source
This commit is contained in:
Devaev Maxim 2020-03-27 03:26:26 +03:00
parent eb923a6f6f
commit d4ff811e1d
4 changed files with 15 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
kvmd/apps/__init__.py
View File

@ -198,7 +198,7 @@ def _get_config_scheme() -> Dict:
}, },
"auth": { "auth": {
"disabled": Option(False, type=valid_bool), "enabled": Option(True, type=valid_bool),
"internal": { "internal": {
"type": Option("htpasswd"), "type": Option("htpasswd"),

2
kvmd/apps/kvmd/__init__.py
View File

@ -69,7 +69,7 @@ def main(argv: Optional[List[str]]=None) -> None:
external_type=config.auth.external.type, external_type=config.auth.external.type,
external_kwargs=(config.auth.external._unpack(ignore=["type"]) if config.auth.external.type else {}), external_kwargs=(config.auth.external._unpack(ignore=["type"]) if config.auth.external.type else {}),
force_internal_users=config.auth.internal.force_users, force_internal_users=config.auth.internal.force_users,
disabled=config.auth.disabled, enabled=config.auth.enabled,
), ),
info_manager=InfoManager(**config.info._unpack()), info_manager=InfoManager(**config.info._unpack()),
log_reader=LogReader(), log_reader=LogReader(),

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

@ -46,20 +46,20 @@ class AuthManager:
external_kwargs: Dict, external_kwargs: Dict,
force_internal_users: List[str], force_internal_users: List[str],
disabled: bool, enabled: bool,
) -> None: ) -> None:
self.__disabled = disabled self.__enabled = enabled
if disabled: if not enabled:
get_logger().warning("AUTHORIZATION IS DISABLED") get_logger().warning("AUTHORIZATION IS DISABLED")
self.__internal_service: Optional[BaseAuthService] = None self.__internal_service: Optional[BaseAuthService] = None
if not disabled: if enabled:
self.__internal_service = get_auth_service_class(internal_type)(**internal_kwargs) self.__internal_service = get_auth_service_class(internal_type)(**internal_kwargs)
get_logger().info("Using internal auth service %r", self.__internal_service.get_plugin_name()) get_logger().info("Using internal auth service %r", self.__internal_service.get_plugin_name())
self.__external_service: Optional[BaseAuthService] = None self.__external_service: Optional[BaseAuthService] = None
if not disabled and external_type: if enabled and external_type:
self.__external_service = get_auth_service_class(external_type)(**external_kwargs) self.__external_service = get_auth_service_class(external_type)(**external_kwargs)
get_logger().info("Using external auth service %r", self.__external_service.get_plugin_name()) get_logger().info("Using external auth service %r", self.__external_service.get_plugin_name())
@ -68,10 +68,10 @@ class AuthManager:
self.__tokens: Dict[str, str] = {} # {token: user} self.__tokens: Dict[str, str] = {} # {token: user}
def is_auth_enabled(self) -> bool: def is_auth_enabled(self) -> bool:
return (not self.__disabled) return self.__enabled
async def authorize(self, user: str, passwd: str) -> bool: async def authorize(self, user: str, passwd: str) -> bool:
assert not self.__disabled assert self.__enabled
assert self.__internal_service assert self.__internal_service
if user not in self.__force_internal_users and self.__external_service: if user not in self.__force_internal_users and self.__external_service:
@ -87,7 +87,7 @@ class AuthManager:
return ok return ok
async def login(self, user: str, passwd: str) -> Optional[str]: async def login(self, user: str, passwd: str) -> Optional[str]:
assert not self.__disabled assert self.__enabled
if (await self.authorize(user, passwd)): if (await self.authorize(user, passwd)):
for (token, token_user) in self.__tokens.items(): for (token, token_user) in self.__tokens.items():
if user == token_user: if user == token_user:
@ -100,18 +100,18 @@ class AuthManager:
return None return None
def logout(self, token: str) -> None: def logout(self, token: str) -> None:
assert not self.__disabled assert self.__enabled
user = self.__tokens.pop(token, "") user = self.__tokens.pop(token, "")
if user: if user:
get_logger().info("Logged out user %r", user) get_logger().info("Logged out user %r", user)
def check(self, token: str) -> Optional[str]: def check(self, token: str) -> Optional[str]:
assert not self.__disabled assert self.__enabled
return self.__tokens.get(token) return self.__tokens.get(token)
@aiotools.atomic @aiotools.atomic
async def cleanup(self) -> None: async def cleanup(self) -> None:
if not self.__disabled: if self.__enabled:
assert self.__internal_service assert self.__internal_service
await self.__internal_service.cleanup() await self.__internal_service.cleanup()
if self.__external_service: if self.__external_service:

4
testenv/tests/apps/kvmd/test_auth.py
View File

@ -59,7 +59,7 @@ async def _get_configured_manager(
external_type=("htpasswd" if external_path else ""), external_type=("htpasswd" if external_path else ""),
external_kwargs=(_make_service_kwargs(external_path) if external_path else {}), external_kwargs=(_make_service_kwargs(external_path) if external_path else {}),
force_internal_users=(force_internal_users or []), force_internal_users=(force_internal_users or []),
disabled=False, enabled=True,
) )
try: try:
@ -147,7 +147,7 @@ async def test_ok__disabled() -> None:
external_type="", external_type="",
external_kwargs={}, external_kwargs={},
force_internal_users=[], force_internal_users=[],
disabled=True, enabled=False,
) )
assert not manager.is_auth_enabled() assert not manager.is_auth_enabled()