option to disable auth

kvmd/apps/__init__.py

@@ -198,6 +198,8 @@ def _get_config_scheme() -> Dict:
             },
 
             "auth": {
+                "disabled": Option(False, type=valid_bool),
+
                 "internal": {
                     "type":  Option("htpasswd"),
                     "force_users": Option([], type=valid_users_list),
@@ -206,6 +208,7 @@ def _get_config_scheme() -> Dict:
 
                 "external": {
                     "type": Option("", type=valid_stripped_string),
+                    # Dynamic content
                 },
             },
 

kvmd/apps/kvmd/__init__.py

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

kvmd/apps/kvmd/auth.py

@@ -46,13 +46,20 @@ class AuthManager:
         external_kwargs: Dict,
 
         force_internal_users: List[str],
+        disabled: bool,
     ) -> None:
 
+        self.__disabled = disabled
+        if disabled:
+            get_logger().warning("AUTHORIZATION IS DISABLED")
+
+        self.__internal_service: Optional[BaseAuthService] = None
+        if not disabled:
             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())
 
         self.__external_service: Optional[BaseAuthService] = None
-        if external_type:
+        if not disabled and external_type:
             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())
 
@@ -60,7 +67,13 @@ class AuthManager:
 
         self.__tokens: Dict[str, str] = {}  # {token: user}
 
+    def is_auth_enabled(self) -> bool:
+        return (not self.__disabled)
+
     async def authorize(self, user: str, passwd: str) -> bool:
+        assert not self.__disabled
+        assert self.__internal_service
+
         if user not in self.__force_internal_users and self.__external_service:
             service = self.__external_service
         else:
@@ -74,6 +87,7 @@ class AuthManager:
         return ok
 
     async def login(self, user: str, passwd: str) -> Optional[str]:
+        assert not self.__disabled
         if (await self.authorize(user, passwd)):
             for (token, token_user) in self.__tokens.items():
                 if user == token_user:
@@ -86,15 +100,19 @@ class AuthManager:
             return None
 
     def logout(self, token: str) -> None:
+        assert not self.__disabled
         user = self.__tokens.pop(token, "")
         if user:
             get_logger().info("Logged out user %r", user)
 
     def check(self, token: str) -> Optional[str]:
+        assert not self.__disabled
         return self.__tokens.get(token)
 
     @aiotools.atomic
     async def cleanup(self) -> None:
+        if not self.__disabled:
+            assert self.__internal_service
             await self.__internal_service.cleanup()
             if self.__external_service:
                 await self.__external_service.cleanup()

kvmd/apps/kvmd/server.py

@@ -167,6 +167,7 @@ class KvmdServer(HttpServer):  # pylint: disable=too-many-arguments,too-many-ins
 
     @exposed_http("POST", "/auth/login", auth_required=False)
     async def __auth_login_handler(self, request: aiohttp.web.Request) -> aiohttp.web.Response:
+        if self.__auth_manager.is_auth_enabled():
             credentials = await request.post()
             token = await self.__auth_manager.login(
                 user=valid_user(credentials.get("user", "")),
@@ -175,9 +176,11 @@ class KvmdServer(HttpServer):  # pylint: disable=too-many-arguments,too-many-ins
             if token:
                 return make_json_response({}, set_cookies={_COOKIE_AUTH_TOKEN: token})
             raise ForbiddenError("Forbidden")
+        return make_json_response({})
 
     @exposed_http("POST", "/auth/logout")
     async def __auth_logout_handler(self, request: aiohttp.web.Request) -> aiohttp.web.Response:
+        if self.__auth_manager.is_auth_enabled():
             token = valid_auth_token(request.cookies.get(_COOKIE_AUTH_TOKEN, ""))
             self.__auth_manager.logout(token)
         return make_json_response({})
@@ -295,7 +298,7 @@ class KvmdServer(HttpServer):  # pylint: disable=too-many-arguments,too-many-ins
     def __add_app_route(self, app: aiohttp.web.Application, exposed: HttpExposed) -> None:
         async def wrapper(request: aiohttp.web.Request) -> aiohttp.web.Response:
             try:
-                if exposed.auth_required:
+                if exposed.auth_required and self.__auth_manager.is_auth_enabled():
                     user = request.headers.get(_HEADER_AUTH_USER, "")
                     passwd = request.headers.get(_HEADER_AUTH_PASSWD, "")
                     token = request.cookies.get(_COOKIE_AUTH_TOKEN, "")

testenv/tests/apps/kvmd/test_auth.py

@@ -59,6 +59,7 @@ async def _get_configured_manager(
         external_type=("htpasswd" if external_path else ""),
         external_kwargs=(_make_service_kwargs(external_path) if external_path else {}),
         force_internal_users=(force_internal_users or []),
+        disabled=False,
     )
 
     try:
@@ -77,6 +78,8 @@ async def test_ok__internal(tmpdir) -> None:  # type: ignore
     htpasswd.save()
 
     async with _get_configured_manager(path) as manager:
+        assert manager.is_auth_enabled()
+
         assert manager.check("xxx") is None
         manager.logout("xxx")
 
@@ -115,6 +118,8 @@ async def test_ok__external(tmpdir) -> None:  # type: ignore
     htpasswd2.save()
 
     async with _get_configured_manager(path1, path2, ["admin"]) as manager:
+        assert manager.is_auth_enabled()
+
         assert (await manager.login("local", "foobar")) is None
         assert (await manager.login("admin", "pass2")) is None
 
@@ -131,3 +136,32 @@ async def test_ok__external(tmpdir) -> None:  # type: ignore
         assert manager.check(token) == "user"
         manager.logout(token)
         assert manager.check(token) is None
+
+
+@pytest.mark.asyncio
+async def test_ok__disabled() -> None:
+    try:
+        manager = AuthManager(
+            internal_type="foobar",
+            internal_kwargs={},
+            external_type="",
+            external_kwargs={},
+            force_internal_users=[],
+            disabled=True,
+        )
+
+        assert not manager.is_auth_enabled()
+
+        with pytest.raises(AssertionError):
+            await manager.authorize("admin", "admin")
+
+        with pytest.raises(AssertionError):
+            await manager.login("admin", "admin")
+
+        with pytest.raises(AssertionError):
+            manager.logout("xxx")
+
+        with pytest.raises(AssertionError):
+            manager.check("xxx")
+    finally:
+        manager.cleanup()