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

auth plugins

Browse Source
This commit is contained in:
Devaev Maxim 2019-04-09 07:13:13 +03:00
parent 0460c2e83b
commit a6028c46a4
16 changed files with 506 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

67
kvmd/apps/__init__.py
View File

@ -35,6 +35,9 @@ import pygments
import pygments.lexers.data
import pygments.formatters
from ..plugins import UnknownPluginError
from ..plugins.auth import get_auth_service_class
from ..yamlconf import ConfigError
from ..yamlconf import make_config
from ..yamlconf import Section
@ -55,8 +58,6 @@ from ..validators.fs import valid_unix_mode
from ..validators.net import valid_ip_or_host
from ..validators.net import valid_port
from ..validators.auth import valid_auth_type
from ..validators.kvm import valid_stream_quality
from ..validators.kvm import valid_stream_fps
@ -84,29 +85,11 @@ def init(
args_parser.add_argument("-m", "--dump-config", dest="dump_config", action="store_true",
help="View current configuration (include all overrides)")
(options, remaining) = args_parser.parse_known_args(argv)
raw_config: Dict = {}
if options.config_path:
options.config_path = os.path.expanduser(options.config_path)
raw_config = load_yaml_file(options.config_path)
scheme = _get_config_scheme()
try:
_merge_dicts(raw_config, build_raw_from_options(options.set_options))
config = make_config(raw_config, scheme)
except ConfigError as err:
raise SystemExit("Config error: " + str(err))
config = _init_config(options.config_path, options.set_options)
if options.dump_config:
dump = make_config_dump(config)
if sys.stdout.isatty():
dump = pygments.highlight(
dump,
pygments.lexers.data.YamlLexer(),
pygments.formatters.TerminalFormatter(bg="dark"), # pylint: disable=no-member
)
print(dump)
sys.exit(0)
_dump_config(config)
raise SystemExit()
logging.captureWarnings(True)
logging.config.dictConfig(config.logging)
@ -114,6 +97,35 @@ def init(
# =====
def _init_config(config_path: str, options: List[str]) -> Section:
config_path = os.path.expanduser(config_path)
raw_config: Dict = load_yaml_file(config_path)
scheme = _get_config_scheme()
try:
_merge_dicts(raw_config, build_raw_from_options(options))
config = make_config(raw_config, scheme)
scheme["kvmd"]["auth"]["internal"] = get_auth_service_class(config.kvmd.auth.internal_type).get_options()
if config.kvmd.auth.external_type:
scheme["kvmd"]["auth"]["external"] = get_auth_service_class(config.kvmd.auth.external_type).get_options()
return make_config(raw_config, scheme)
except (ConfigError, UnknownPluginError) as err:
raise SystemExit("Config error: %s" % (str(err)))
def _dump_config(config: Section) -> None:
dump = make_config_dump(config)
if sys.stdout.isatty():
dump = pygments.highlight(
dump,
pygments.lexers.data.YamlLexer(),
pygments.formatters.TerminalFormatter(bg="dark"), # pylint: disable=no-member
)
print(dump)
def _merge_dicts(dest: Dict, src: Dict) -> None:
for key in src:
if key in dest:
@ -138,10 +150,11 @@ def _get_config_scheme() -> Dict:
},
"auth": {
"type": Option("htpasswd", type=valid_auth_type, unpack_as="auth_type"),
"htpasswd": {
"file": Option("/etc/kvmd/htpasswd", type=valid_abs_path_exists, unpack_as="path"),
},
"internal_users": Option([]),
"internal_type": Option("htpasswd"),
"external_type": Option(""),
# "internal": {},
# "external": {},
},
"info": {

6
kvmd/apps/htpasswd/__init__.py
View File

@ -44,9 +44,9 @@ from .. import init
# =====
def _get_htpasswd_path(config: Section) -> str:
if config.kvmd.auth.type != "htpasswd":
print("Warning: KVMD does not use htpasswd auth", file=sys.stderr)
return config.kvmd.auth.htpasswd.file
if config.kvmd.auth.internal_type != "htpasswd":
raise SystemExit("Error: KVMD internal auth not using 'htpasswd' (now configured %r)" % (config.kvmd.auth.internal_type))
return config.kvmd.auth.internal.file
@contextlib.contextmanager

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

@ -22,6 +22,9 @@
import asyncio
from typing import List
from typing import Optional
from ...logging import get_logger
from ... import gpio
@ -39,13 +42,19 @@ from .server import Server
# =====
def main() -> None:
config = init("kvmd", description="The main Pi-KVM daemon")[2].kvmd
def main(argv: Optional[List[str]]=None) -> None:
config = init("kvmd", description="The main Pi-KVM daemon", argv=argv)[2].kvmd
with gpio.bcm():
# pylint: disable=protected-access
loop = asyncio.get_event_loop()
Server(
auth_manager=AuthManager(**config.auth._unpack()),
auth_manager=AuthManager(
internal_users=config.auth.internal_users,
internal_type=config.auth.internal_type,
external_type=config.auth.external_type,
internal=config.auth.internal._unpack(),
external=(config.auth.external._unpack() if config.auth.external_type else {}),
),
info_manager=InfoManager(loop=loop, **config.info._unpack()),
log_reader=LogReader(loop=loop),

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

@ -22,33 +22,56 @@
import secrets
from typing import List
from typing import Dict
from typing import Optional
import passlib.apache
from ...logging import get_logger
from ...plugins.auth import BaseAuthService
from ...plugins.auth import get_auth_service_class
# =====
class AuthManager:
def __init__(self, auth_type: str, htpasswd: Dict) -> None:
self.__login = {
"htpasswd": lambda: _HtpasswdLogin(**htpasswd),
}[auth_type]().login
def __init__(
self,
internal_users: List[str],
internal_type: str,
external_type: str,
internal: Dict,
external: Dict,
) -> None:
self.__internal_users = internal_users
self.__internal_service = get_auth_service_class(internal_type)(**internal)
get_logger().info("Using internal login service %r", self.__internal_service.PLUGIN_NAME)
self.__external_service: Optional[BaseAuthService] = None
if external_type:
self.__external_service = get_auth_service_class(external_type)(**external)
get_logger().info("Using external login service %r", self.__external_service.PLUGIN_NAME)
self.__tokens: Dict[str, str] = {} # {token: user}
def login(self, user: str, passwd: str) -> Optional[str]:
if self.__login(user, passwd):
async def login(self, user: str, passwd: str) -> Optional[str]:
if user not in self.__internal_users and self.__external_service:
service = self.__external_service
else:
service = self.__internal_service
if (await service.login(user, passwd)):
for (token, token_user) in self.__tokens.items():
if user == token_user:
return token
token = secrets.token_hex(32)
self.__tokens[token] = user
get_logger().info("Logged in user %r", user)
get_logger().info("Logged in user %r via login service %r", user, service.PLUGIN_NAME)
return token
else:
get_logger().error("Access denied for user %r", user)
get_logger().error("Access denied for user %r from login service %r", user, service.PLUGIN_NAME)
return None
def logout(self, token: str) -> None:
@ -59,12 +82,7 @@ class AuthManager:
def check(self, token: str) -> Optional[str]:
return self.__tokens.get(token)
class _HtpasswdLogin:
def __init__(self, path: str) -> None:
get_logger().info("Using htpasswd auth file %r", path)
self.__path = path
def login(self, user: str, passwd: str) -> bool:
htpasswd = passlib.apache.HtpasswdFile(self.__path)
return htpasswd.check_password(user, passwd)
async def cleanup(self) -> None:
await self.__internal_service.cleanup()
if self.__external_service:
await self.__external_service.cleanup()

17
kvmd/apps/kvmd/server.py
View File

@ -311,7 +311,7 @@ class Server: # pylint: disable=too-many-instance-attributes
@_exposed("POST", "/auth/login", auth_required=False)
async def __auth_login_handler(self, request: aiohttp.web.Request) -> aiohttp.web.Response:
credentials = await request.post()
token = self._auth_manager.login(
token = await self._auth_manager.login(
user=valid_user(credentials.get("user", "")),
passwd=valid_passwd(credentials.get("passwd", "")),
)
@ -533,9 +533,18 @@ class Server: # pylint: disable=too-many-instance-attributes
await self.__remove_socket(ws)
async def __on_cleanup(self, _: aiohttp.web.Application) -> None:
await self.__streamer.cleanup()
await self.__msd.cleanup()
await self.__hid.cleanup()
logger = get_logger(0)
for obj in [
self._auth_manager,
self.__streamer,
self.__msd,
self.__hid,
]:
logger.info("Cleaning up %s ...", type(obj).__name__)
try:
await obj.cleanup() # type: ignore
except Exception:
logger.exception("Cleanup error")
async def __broadcast_event(self, event_type: _Events, event_attrs: Dict) -> None:
if self.__sockets:

71
kvmd/plugins/__init__.py Normal file
View File

@ -0,0 +1,71 @@
# ========================================================================== #
# #
# KVMD - The main Pi-KVM daemon. #
# #
# Copyright (C) 2018 Maxim Devaev <mdevaev@gmail.com> #
# #
# This program is free software: you can redistribute it and/or modify #
# it under the terms of the GNU General Public License as published by #
# the Free Software Foundation, either version 3 of the License, or #
# (at your option) any later version. #
# #
# This program is distributed in the hope that it will be useful, #
# but WITHOUT ANY WARRANTY; without even the implied warranty of #
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
# GNU General Public License for more details. #
# #
# You should have received a copy of the GNU General Public License #
# along with this program. If not, see <https://www.gnu.org/licenses/>. #
# #
# ========================================================================== #
import importlib
import functools
import os
from typing import Dict
from typing import Type
from typing import Any
from ..yamlconf import Option
# =====
class UnknownPluginError(Exception):
pass
# =====
class BasePlugin:
PLUGIN_NAME: str = ""
def __init__(self, **_: Any) -> None:
pass
@classmethod
def get_options(cls) -> Dict[str, Option]:
return {}
# =====
def get_plugin_class(sub: str, name: str) -> Type[BasePlugin]:
classes = _get_plugin_classes(sub)
try:
return classes[name]
except KeyError:
raise UnknownPluginError("Unknown plugin '%s/%s'" % (sub, name))
# =====
@functools.lru_cache()
def _get_plugin_classes(sub: str) -> Dict[str, Type[BasePlugin]]:
classes: Dict[str, Type[BasePlugin]] = {} # noqa: E701
sub_path = os.path.join(os.path.dirname(__file__), sub)
for file_name in os.listdir(sub_path):
if not file_name.startswith("__") and file_name.endswith(".py"):
module_name = file_name[:-3]
module = importlib.import_module("kvmd.plugins.{}.{}".format(sub, module_name))
plugin_class = getattr(module, "Plugin")
classes[plugin_class.PLUGIN_NAME] = plugin_class
return classes

40
kvmd/plugins/auth/__init__.py Normal file
View File

@ -0,0 +1,40 @@
# ========================================================================== #
# #
# KVMD - The main Pi-KVM daemon. #
# #
# Copyright (C) 2018 Maxim Devaev <mdevaev@gmail.com> #
# #
# This program is free software: you can redistribute it and/or modify #
# it under the terms of the GNU General Public License as published by #
# the Free Software Foundation, either version 3 of the License, or #
# (at your option) any later version. #
# #
# This program is distributed in the hope that it will be useful, #
# but WITHOUT ANY WARRANTY; without even the implied warranty of #
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
# GNU General Public License for more details. #
# #
# You should have received a copy of the GNU General Public License #
# along with this program. If not, see <https://www.gnu.org/licenses/>. #
# #
# ========================================================================== #
from typing import Type
from .. import BasePlugin
from .. import get_plugin_class
# =====
class BaseAuthService(BasePlugin):
async def login(self, user: str, passwd: str) -> bool:
raise NotImplementedError
async def cleanup(self) -> None:
pass
# =====
def get_auth_service_class(name: str) -> Type[BaseAuthService]:
return get_plugin_class("auth", name) # type: ignore

49
kvmd/plugins/auth/htpasswd.py Normal file
View File

@ -0,0 +1,49 @@
# ========================================================================== #
# #
# KVMD - The main Pi-KVM daemon. #
# #
# Copyright (C) 2018 Maxim Devaev <mdevaev@gmail.com> #
# #
# This program is free software: you can redistribute it and/or modify #
# it under the terms of the GNU General Public License as published by #
# the Free Software Foundation, either version 3 of the License, or #
# (at your option) any later version. #
# #
# This program is distributed in the hope that it will be useful, #
# but WITHOUT ANY WARRANTY; without even the implied warranty of #
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
# GNU General Public License for more details. #
# #
# You should have received a copy of the GNU General Public License #
# along with this program. If not, see <https://www.gnu.org/licenses/>. #
# #
# ========================================================================== #
from typing import Dict
import passlib.apache
from ...yamlconf import Option
from ...validators.fs import valid_abs_path_exists
from . import BaseAuthService
# =====
class Plugin(BaseAuthService):
PLUGIN_NAME = "htpasswd"
def __init__(self, path: str) -> None: # pylint: disable=super-init-not-called
self.__path = path
@classmethod
def get_options(cls) -> Dict[str, Option]:
return {
"file": Option("/etc/kvmd/htpasswd", type=valid_abs_path_exists, unpack_as="path"),
}
async def login(self, user: str, passwd: str) -> bool:
htpasswd = passlib.apache.HtpasswdFile(self.__path)
return htpasswd.check_password(user, passwd)

111
kvmd/plugins/auth/http.py Normal file
View File

@ -0,0 +1,111 @@
# ========================================================================== #
# #
# KVMD - The main Pi-KVM daemon. #
# #
# Copyright (C) 2018 Maxim Devaev <mdevaev@gmail.com> #
# #
# This program is free software: you can redistribute it and/or modify #
# it under the terms of the GNU General Public License as published by #
# the Free Software Foundation, either version 3 of the License, or #
# (at your option) any later version. #
# #
# This program is distributed in the hope that it will be useful, #
# but WITHOUT ANY WARRANTY; without even the implied warranty of #
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
# GNU General Public License for more details. #
# #
# You should have received a copy of the GNU General Public License #
# along with this program. If not, see <https://www.gnu.org/licenses/>. #
# #
# ========================================================================== #
from typing import Dict
from typing import Optional
import aiohttp
import aiohttp.web
from ...yamlconf import Option
from ...validators.basic import valid_bool
from ...validators.basic import valid_float_f01
from ...logging import get_logger
from ... import __version__
from . import BaseAuthService
# =====
class Plugin(BaseAuthService):
PLUGIN_NAME = "http"
def __init__( # pylint: disable=super-init-not-called
self,
url: str,
verify: bool,
post: bool,
user: str,
passwd: str,
timeout: float,
) -> None:
self.__url = url
self.__verify = verify
self.__post = post
self.__user = user
self.__passwd = passwd
self.__timeout = timeout
self.__http_session: Optional[aiohttp.ClientSession] = None
@classmethod
def get_options(cls) -> Dict[str, Option]:
return {
"url": Option("http://localhost/auth_post"),
"verify": Option(True, type=valid_bool),
"post": Option(True, type=valid_bool),
"user": Option(""),
"passwd": Option(""),
"timeout": Option(5.0, type=valid_float_f01),
}
async def login(self, user: str, passwd: str) -> bool:
kwargs: Dict = {
"method": "GET",
"url": self.__url,
"timeout": self.__timeout,
"headers": {
"User-Agent": "KVMD/%s" % (__version__),
"X-KVMD-User": user,
},
}
if self.__post:
kwargs["method"] = "POST"
kwargs["json"] = {"user": user, "passwd": passwd}
session = self.__ensure_session()
try:
async with session.request(**kwargs) as response:
response.raise_for_status()
return True
except Exception:
get_logger().exception("Failed HTTP auth request for user %r", user)
return False
async def cleanup(self) -> None:
if self.__http_session:
await self.__http_session.close()
self.__http_session = None
def __ensure_session(self) -> aiohttp.ClientSession:
if not self.__http_session:
kwargs: Dict = {}
if self.__user:
kwargs["auth"] = aiohttp.BasicAuth(login=self.__user, password=self.__passwd)
if not self.__verify:
kwargs["connector"] = aiohttp.TCPConnector(ssl=False)
self.__http_session = aiohttp.ClientSession(**kwargs)
return self.__http_session

5
kvmd/validators/auth.py
View File

@ -22,7 +22,6 @@
from typing import Any
from . import check_string_in_list
from . import check_re_match
@ -37,7 +36,3 @@ def valid_passwd(arg: Any) -> str:
def valid_auth_token(arg: Any) -> str:
return check_re_match(arg, "auth token", r"^[0-9a-f]{64}$", hide=True)
def valid_auth_type(arg: Any) -> str:
return check_string_in_list(arg, "auth type", ["htpasswd"])

2
setup.py
View File

@ -40,6 +40,8 @@ def main() -> None:
"kvmd",
"kvmd.validators",
"kvmd.yamlconf",
"kvmd.plugins",
"kvmd.plugins.auth",
"kvmd.apps",
"kvmd.apps.kvmd",
"kvmd.apps.htpasswd",

5
testenv/tox.ini
View File

@ -28,7 +28,7 @@ deps =
-rrequirements.txt
[testenv:vulture]
commands = vulture --ignore-names=_format_P --ignore-decorators=@_exposed,@_system_task,@pytest.fixture kvmd genmap.py tests testenv/linters/vulture-wl.py
commands = vulture --ignore-names=_format_P,Plugin --ignore-decorators=@_exposed,@_system_task,@pytest.fixture kvmd genmap.py tests testenv/linters/vulture-wl.py
deps =
vulture
-rrequirements.txt
@ -38,8 +38,9 @@ commands = py.test -vv --cov-config=testenv/linters/coverage.ini --cov-report=te
deps =
pytest
pytest-cov
pytest-asyncio
pytest-mock
pytest-asyncio
pytest-aiohttp
-rrequirements.txt
[testenv:eslint]

2
tests/test_app_htpasswd.py
View File

@ -59,7 +59,7 @@ def _run_htpasswd(htpasswd: passlib.apache.HtpasswdFile, cmd: List[str]) -> None
"kvmd-htpasswd",
*cmd,
"--set-options",
"kvmd/auth/htpasswd/file=" + htpasswd.path,
"kvmd/auth/internal/file=" + htpasswd.path,
])

55
tests/test_auth_service_htpasswd.py Normal file
View File

@ -0,0 +1,55 @@
# ========================================================================== #
# #
# KVMD - The main Pi-KVM daemon. #
# #
# Copyright (C) 2018 Maxim Devaev <mdevaev@gmail.com> #
# #
# This program is free software: you can redistribute it and/or modify #
# it under the terms of the GNU General Public License as published by #
# the Free Software Foundation, either version 3 of the License, or #
# (at your option) any later version. #
# #
# This program is distributed in the hope that it will be useful, #
# but WITHOUT ANY WARRANTY; without even the implied warranty of #
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
# GNU General Public License for more details. #
# #
# You should have received a copy of the GNU General Public License #
# along with this program. If not, see <https://www.gnu.org/licenses/>. #
# #
# ========================================================================== #
import os
import passlib.apache
import pytest
from kvmd.plugins.auth import get_auth_service_class
# =====
@pytest.mark.asyncio
async def test_ok__htpasswd_service(tmpdir) -> None: # type: ignore
path = os.path.abspath(str(tmpdir.join("htpasswd")))
htpasswd = passlib.apache.HtpasswdFile(path, new=True)
htpasswd.set_password("admin", "foo")
htpasswd.save()
service = get_auth_service_class("htpasswd")(path=path)
assert (await service.login("admin", "foo"))
assert not (await service.login("user", "foo"))
htpasswd.set_password("admin", "bar")
htpasswd.set_password("user", "bar")
htpasswd.save()
assert (await service.login("admin", "bar"))
assert (await service.login("user", "bar"))
assert not (await service.login("admin", "foo"))
assert not (await service.login("user", "foo"))
await service.cleanup()

69
tests/test_auth_service_http.py Normal file
View File

@ -0,0 +1,69 @@
# ========================================================================== #
# #
# KVMD - The main Pi-KVM daemon. #
# #
# Copyright (C) 2018 Maxim Devaev <mdevaev@gmail.com> #
# #
# This program is free software: you can redistribute it and/or modify #
# it under the terms of the GNU General Public License as published by #
# the Free Software Foundation, either version 3 of the License, or #
# (at your option) any later version. #
# #
# This program is distributed in the hope that it will be useful, #
# but WITHOUT ANY WARRANTY; without even the implied warranty of #
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
# GNU General Public License for more details. #
# #
# You should have received a copy of the GNU General Public License #
# along with this program. If not, see <https://www.gnu.org/licenses/>. #
# #
# ========================================================================== #
from typing import AsyncGenerator
import aiohttp.web
import pytest
from kvmd.plugins.auth import get_auth_service_class
# =====
async def _handle_auth_post(request: aiohttp.web.BaseRequest) -> aiohttp.web.Response:
status = 400
if request.method == "POST":
credentials = (await request.json())
if credentials["user"] == "admin" and credentials["passwd"] == "foobar":
status = 200
return aiohttp.web.Response(text=str(status), status=status)
@pytest.fixture(name="auth_server_port")
async def _auth_server_port_fixture(aiohttp_server) -> AsyncGenerator[int, None]: # type: ignore
app = aiohttp.web.Application()
app.router.add_post("/auth_post", _handle_auth_post)
server = await aiohttp_server(app)
try:
yield server.port
finally:
await server.close()
# =====
@pytest.mark.asyncio
async def test_ok__http_service(auth_server_port: int) -> None:
service = get_auth_service_class("http")(
url="http://localhost:%d/auth_post" % (auth_server_port),
verify=False,
post=True,
user="",
passwd="",
timeout=5.0,
)
assert not (await service.login("admin", "foo"))
assert not (await service.login("user", "foo"))
assert (await service.login("admin", "foobar"))
await service.cleanup()

12
tests/test_validators_auth.py
View File

@ -28,7 +28,6 @@ from kvmd.validators import ValidatorError
from kvmd.validators.auth import valid_user
from kvmd.validators.auth import valid_passwd
from kvmd.validators.auth import valid_auth_token
from kvmd.validators.auth import valid_auth_type
# =====
@ -106,14 +105,3 @@ def test_ok__valid_auth_token(arg: Any) -> None:
def test_fail__valid_auth_token(arg: Any) -> None:
with pytest.raises(ValidatorError):
print(valid_auth_token(arg))
@pytest.mark.parametrize("arg", ["HTPASSWD ", "htpasswd"])
def test_ok__valid_auth_type(arg: Any) -> None:
assert valid_auth_type(arg) == arg.strip().lower()
@pytest.mark.parametrize("arg", ["test", "", None])
def test_fail__valid_auth_type(arg: Any) -> None:
with pytest.raises(ValidatorError):
print(valid_auth_type(arg))