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

wake-on-lan back

Browse Source
This commit is contained in:
Devaev Maxim 2019-11-29 01:35:38 +03:00
parent 51e15d01c2
commit 3d8f16b9c6
7 changed files with 155 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
configs/os/services/kvmd.service
View File

@ -8,6 +8,7 @@ Group=kvmd
Type=simple
Restart=always
RestartSec=3
AmbientCapabilities=CAP_NET_RAW
ExecStart=/usr/bin/kvmd
ExecStopPost=/usr/bin/kvmd-cleanup

8
kvmd/apps/__init__.py
View File

@ -66,7 +66,9 @@ from ..validators.os import valid_unix_mode
from ..validators.os import valid_command
from ..validators.net import valid_ip_or_host
from ..validators.net import valid_ip
from ..validators.net import valid_port
from ..validators.net import valid_mac
from ..validators.kvm import valid_stream_quality
from ..validators.kvm import valid_stream_fps
@ -212,6 +214,12 @@ def _get_config_scheme() -> Dict:
"extras": Option("/usr/share/kvmd/extras", type=valid_abs_dir, unpack_as="extras_path"),
},
"wol": {
"ip": Option("255.255.255.255", type=(lambda arg: valid_ip(arg, v6=False))),
"port": Option(9, type=valid_port),
"mac": Option("", type=(lambda arg: (valid_mac(arg) if arg else ""))),
},
"hid": {
"type": Option("", type=valid_stripped_string_not_empty),
# Dynamic content

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

@ -36,6 +36,7 @@ from .. import init
from .auth import AuthManager
from .info import InfoManager
from .logreader import LogReader
from .wol import WakeOnLan
from .streamer import Streamer
from .server import Server
@ -71,6 +72,7 @@ def main(argv: Optional[List[str]]=None) -> None:
),
info_manager=InfoManager(**config.info._unpack()),
log_reader=LogReader(),
wol=WakeOnLan(**config.wol._unpack()),
hid=get_hid_class(config.hid.type)(**config.hid._unpack(ignore=["type"])),
atx=get_atx_class(config.atx.type)(**config.atx._unpack(ignore=["type"])),

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

@ -82,6 +82,9 @@ from .info import InfoManager
from .logreader import LogReader
from .streamer import Streamer
from .wol import WolDisabledError
from .wol import WakeOnLan
# =====
try:
@ -191,7 +194,7 @@ def _exposed(http_method: str, path: str, auth_required: bool=True) -> Callable:
except (AtxIsBusyError, MsdIsBusyError) as err:
return _json_exception(err, 409)
except (ValidatorError, AtxOperationError, MsdOperationError) as err:
except (ValidatorError, AtxOperationError, MsdOperationError, WolDisabledError) as err:
return _json_exception(err, 400)
except UnauthorizedError as err:
return _json_exception(err, 401)
@ -222,6 +225,7 @@ def _system_task(method: Callable) -> Callable:
class _Events(Enum):
INFO_STATE = "info_state"
WOL_STATE = "wol_state"
HID_STATE = "hid_state"
ATX_STATE = "atx_state"
MSD_STATE = "msd_state"
@ -234,6 +238,7 @@ class Server: # pylint: disable=too-many-instance-attributes
auth_manager: AuthManager,
info_manager: InfoManager,
log_reader: LogReader,
wol: WakeOnLan,
hid: BaseHid,
atx: BaseAtx,
@ -244,6 +249,7 @@ class Server: # pylint: disable=too-many-instance-attributes
self._auth_manager = auth_manager
self.__info_manager = info_manager
self.__log_reader = log_reader
self.__wol = wol
self.__hid = hid
self.__atx = atx
@ -355,6 +361,17 @@ class Server: # pylint: disable=too-many-instance-attributes
)).encode("utf-8") + b"\r\n")
return response
# ===== Wake-on-LAN
@_exposed("GET", "/wol")
async def __wol_state_handler(self, _: aiohttp.web.Request) -> aiohttp.web.Response:
return _json(self.__wol.get_state())
@_exposed("POST", "/wol/wakeup")
async def __wol_wakeup_handler(self, _: aiohttp.web.Request) -> aiohttp.web.Response:
await self.__wol.wakeup()
return _json()
# ===== WEBSOCKET
@_exposed("GET", "/ws")
@ -366,6 +383,7 @@ class Server: # pylint: disable=too-many-instance-attributes
await self.__register_socket(ws)
await asyncio.gather(*[
self.__broadcast_event(_Events.INFO_STATE, (await self.__make_info())),
self.__broadcast_event(_Events.WOL_STATE, self.__wol.get_state()),
self.__broadcast_event(_Events.HID_STATE, self.__hid.get_state()),
self.__broadcast_event(_Events.ATX_STATE, self.__atx.get_state()),
self.__broadcast_event(_Events.MSD_STATE, (await self.__msd.get_state())),

88
kvmd/apps/kvmd/wol.py Normal file
View File

@ -0,0 +1,88 @@
# ========================================================================== #
# #
# 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 socket
from typing import Dict
from typing import Optional
from ...logging import get_logger
from ... import aiotools
# =====
class WolDisabledError(Exception):
def __init__(self) -> None:
super().__init__("WoL is disabled")
# =====
class WakeOnLan:
def __init__(self, ip: str, port: int, mac: str) -> None:
self.__ip = ip
self.__port = port
self.__mac = mac
self.__magic = b""
if mac:
assert len(mac) == 17, mac
self.__magic = bytes.fromhex("FF" * 6 + mac.replace(":", "") * 16)
def get_state(self) -> Dict:
return {
"enabled": bool(self.__magic),
"target": {
"ip": self.__ip,
"port": self.__port,
"mac": self.__mac,
},
}
@aiotools.atomic
async def wakeup(self) -> None:
if not self.__magic:
raise WolDisabledError()
await self.__inner_wakeup()
@aiotools.tasked
@aiotools.muted("Can't perform Wake-on-LAN or operation was not completed")
async def __inner_wakeup(self) -> None:
logger = get_logger(0)
logger.info("Waking up %s (%s:%s) using Wake-on-LAN ...", self.__mac, self.__ip, self.__port)
sock: Optional[socket.socket] = None
try:
# TODO: IPv6 support: http://lists.cluenet.de/pipermail/ipv6-ops/2014-September/010139.html
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
sock.connect((self.__ip, self.__port))
sock.send(self.__magic)
except Exception:
logger.exception("Can't send Wake-on-LAN packet")
else:
logger.info("Wake-on-LAN packet sent")
finally:
if sock:
try:
sock.close()
except Exception:
pass

20
kvmd/validators/net.py
View File

@ -22,6 +22,8 @@
import socket
from typing import List
from typing import Callable
from typing import Any
from . import check_re_match
@ -44,15 +46,18 @@ def valid_ip_or_host(arg: Any) -> str:
)
def valid_ip(arg: Any) -> str:
def valid_ip(arg: Any, v4: bool=True, v6: bool=True) -> str:
assert v4 or v6
validators: List[Callable] = []
if v4:
validators.append(lambda arg: (arg, socket.inet_pton(socket.AF_INET, arg))[0])
if v6:
validators.append(lambda arg: (arg, socket.inet_pton(socket.AF_INET6, arg))[0])
name = "IP address"
return check_any(
arg=valid_stripped_string_not_empty(arg, name),
name=name,
validators=[
lambda arg: (arg, socket.inet_pton(socket.AF_INET, arg))[0],
lambda arg: (arg, socket.inet_pton(socket.AF_INET6, arg))[0],
],
validators=validators,
)
@ -65,3 +70,8 @@ def valid_rfc_host(arg: Any) -> str:
def valid_port(arg: Any) -> int:
return int(valid_number(arg, min=0, max=65535, name="TCP/UDP port"))
def valid_mac(arg: Any) -> str:
pattern = ":".join([r"[0-9a-fA-F]{2}"] * 6)
return check_re_match(arg, "MAC address", pattern).lower()

22
testenv/tests/validators/test_net.py
View File

@ -29,6 +29,7 @@ from kvmd.validators.net import valid_ip_or_host
from kvmd.validators.net import valid_ip
from kvmd.validators.net import valid_rfc_host
from kvmd.validators.net import valid_port
from kvmd.validators.net import valid_mac
# =====
@ -120,3 +121,24 @@ def test_ok__valid_port(arg: Any) -> None:
def test_fail__valid_port(arg: Any) -> None:
with pytest.raises(ValidatorError):
print(valid_port(arg))
# =====
@pytest.mark.parametrize("arg", [
" 00:00:00:00:00:00 ",
" 9f:00:00:00:00:00 ",
" FF:FF:FF:FF:FF:FF ",
])
def test_ok__valid_mac(arg: Any) -> None:
assert valid_mac(arg) == arg.strip().lower()
@pytest.mark.parametrize("arg", [
"00:00:00:00:00:0",
"9x:00:00:00:00:00",
"",
None,
])
def test_fail__valid_mac(arg: Any) -> None:
with pytest.raises(ValidatorError):
print(valid_mac(arg))