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

otg service and helper

Browse Source
This commit is contained in:
Devaev Maxim 2019-09-24 05:50:21 +03:00
parent aee005787c
commit e97a2ea251
9 changed files with 331 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
PKGBUILD
View File

@ -28,6 +28,7 @@ depends=(
python-raspberry-gpio
python-pyserial
python-setproctitle
python-psutil
python-systemd
python-dbus
python-pygments

13
configs/os/services/kvmd-otg.service Normal file
View File

@ -0,0 +1,13 @@
[Unit]
Description=Pi-KVM - OTG setup
After=systemd-modules-load.service
Before=kvmd.service
[Service]
Type=oneshot
ExecStart=/usr/bin/kvmd-otg start
ExecStop=/usr/bin/kvmd-otg stop
RemainAfterExit=true
[Install]
WantedBy=multi-user.target

11
kvmd/apps/__init__.py
View File

@ -241,6 +241,17 @@ def _get_config_scheme() -> Dict:
},
},
"otg": {
"gadget": Option("pikvm"),
"vendor_id": Option(0x1D6B, type=valid_number), # Linux Foundation
"product_id": Option(0x0104, type=valid_number), # Multifunction Composite Gadget
"manufacturer": Option("Pi-KVM"),
"product": Option("Composite KVM Device"),
"serial_number": Option("CAFEBABE"),
"udc": Option(""),
"acm": Option(True, type=valid_bool),
},
"ipmi": {
"server": {
"host": Option("::", type=valid_ip_or_host),

182
kvmd/apps/otg/__init__.py Normal file
View File

@ -0,0 +1,182 @@
# ========================================================================== #
# #
# 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 os import listdir
from os import mkdir
from os import makedirs
from os import symlink
from os import rmdir
from os import unlink
from os.path import join
import argparse
from typing import List
from typing import Optional
from ...yamlconf import Section
from ...validators import ValidatorError
from .. import init
# =====
def _write(path: str, text: str) -> None:
with open(path, "w") as param_file:
param_file.write(text)
def _find_udc(udc: str) -> str:
udcs = sorted(listdir("/sys/class/udc"))
if not udc:
if len(udcs) == 0:
raise RuntimeError("Can't find any UDC")
udc = udcs[0]
elif udc not in udcs:
raise RuntimeError(f"Can't find selected UDC: {udc}")
return udc
def _check_config(config: Section) -> None:
if (
not config.otg.acm
and config.kvmd.hid.type != "otg"
and config.kvmd.msd.type != "otg"
):
raise RuntimeError("Nothing to do")
def _cmd_start(config: Section) -> None:
# https://www.kernel.org/doc/Documentation/usb/gadget_configfs.txt
# https://www.isticktoit.net/?p=1383
_check_config(config)
udc = _find_udc(config.otg.udc)
gadget_path = join("/sys/kernel/config/usb_gadget", config.otg.gadget)
mkdir(gadget_path)
_write(join(gadget_path, "idVendor"), f"0x{config.otg.vendor_id:X}")
_write(join(gadget_path, "idProduct"), f"0x{config.otg.product_id:X}")
_write(join(gadget_path, "bcdDevice"), "0x0100")
_write(join(gadget_path, "bcdUSB"), "0x0200")
lang_path = join(gadget_path, "strings/0x409")
mkdir(lang_path)
_write(join(lang_path, "manufacturer"), config.otg.manufacturer)
_write(join(lang_path, "product"), config.otg.product)
_write(join(lang_path, "serialnumber"), config.otg.serial_number)
config_path = join(gadget_path, "configs/c.1")
makedirs(join(config_path, "strings/0x409"))
_write(join(gadget_path, "configs/c.1/strings/0x409/configuration"), "Config 1: ECM network")
_write(join(gadget_path, "configs/c.1/MaxPower"), "250")
if config.otg.acm:
func_path = join(gadget_path, "functions/acm.usb0")
mkdir(func_path)
symlink(func_path, join(config_path, "acm.usb0"))
if config.kvmd.hid.type == "otg":
func_path = join(gadget_path, "functions/hid.usb0")
mkdir(func_path)
_write(join(func_path, "protocol"), "1")
_write(join(func_path, "subclass"), "1")
_write(join(func_path, "report_length"), "1")
with open(join(func_path, "report_desc"), "wb") as report_file:
report_file.write(
b"\x05\x01\x09\x06\xa1\x01\x05\x07\x19\xe0\x29\xe7\x15\x00"
b"\x25\x01\x75\x01\x95\x08\x81\x02\x95\x01\x75\x08\x81\x03"
b"\x95\x05\x75\x01\x05\x08\x19\x01\x29\x05\x91\x02\x95\x01"
b"\x75\x03\x91\x03\x95\x06\x75\x08\x15\x00\x25\x65\x05\x07"
b"\x19\x00\x29\x65\x81\x00\xc0"
)
symlink(func_path, join(config_path, "hid.usb0"))
if config.kvmd.msd.type == "otg":
func_path = join(gadget_path, "functions/mass_storage.usb0")
mkdir(func_path)
_write(join(func_path, "stall"), "0")
_write(join(func_path, "lun.0/cdrom"), "1")
_write(join(func_path, "lun.0/ro"), "1")
_write(join(func_path, "lun.0/removable"), "1")
_write(join(func_path, "lun.0/nofua"), "0")
symlink(func_path, join(config_path, "mass_storage.usb0"))
_write(join(gadget_path, "UDC"), udc)
def _cmd_stop(config: Section) -> None:
# https://www.kernel.org/doc/Documentation/usb/gadget_configfs.txt
_check_config(config)
gadget_path = join("/sys/kernel/config/usb_gadget", config.otg.gadget)
_write(join(gadget_path, "UDC"), "")
config_path = join(gadget_path, "configs/c.1")
for func in listdir(config_path):
if func.endswith(".usb0"):
unlink(join(config_path, func))
rmdir(join(config_path, "strings/0x409"))
rmdir(config_path)
funcs_path = join(gadget_path, "functions")
for func in listdir(funcs_path):
if func.endswith(".usb0"):
rmdir(join(funcs_path, func))
rmdir(join(gadget_path, "strings/0x409"))
rmdir(gadget_path)
# =====
def main(argv: Optional[List[str]]=None) -> None:
(parent_parser, argv, config) = init(
add_help=False,
argv=argv,
load_hid=True,
load_atx=True,
load_msd=True,
)
parser = argparse.ArgumentParser(
prog="kvmd-otg",
description="Control KVMD OTG device",
parents=[parent_parser],
)
parser.set_defaults(cmd=(lambda *_: parser.print_help()))
subparsers = parser.add_subparsers()
cmd_start_parser = subparsers.add_parser("start", help="Start OTG")
cmd_start_parser.set_defaults(cmd=_cmd_start)
cmd_stop_parser = subparsers.add_parser("stop", help="Stop OTG")
cmd_stop_parser.set_defaults(cmd=_cmd_stop)
options = parser.parse_args(argv[1:])
try:
options.cmd(config)
except ValidatorError as err:
raise SystemExit(str(err))

24
kvmd/apps/otg/__main__.py Normal file
View File

@ -0,0 +1,24 @@
# ========================================================================== #
# #
# 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 . import main
main()

71
kvmd/apps/otgmsd/__init__.py Normal file
View File

@ -0,0 +1,71 @@
#!/usr/bin/env python3
# ========================================================================== #
# #
# 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 signal
import errno
import argparse
import psutil
# =====
def _set_msd_image(gadget: str, path: str) -> None:
lun_file_path = os.path.join("/sys/kernel/config/usb_gadget", gadget, "functions/mass_storage.usb0/lun.0/file")
try:
with open(lun_file_path, "w") as lun_file:
lun_file.write(path + "\n")
except OSError as err:
if err.errno == errno.EBUSY:
raise SystemExit(f"Can't change image because device is locked: {str(err)}")
raise
def _reset_msd() -> None:
# https://github.com/torvalds/linux/blob/3039fadf2bfdc104dc963820c305778c7c1a6229/drivers/usb/gadget/function/f_mass_storage.c#L2924
found = False
for proc in psutil.process_iter():
attrs = proc.as_dict(attrs=["name", "exe", "pid"])
if attrs.get("name") == "file-storage" and not attrs.get("exe"):
try:
proc.send_signal(signal.SIGUSR1)
found = True
except Exception as err:
SystemExit(f"Can't send SIGUSR1 to MSD kernel thread with pid={attrs['pid']}: {str(err)}")
if not found:
raise SystemExit("Can't find MSD kernel thread")
# =====
def main() -> None:
parser = argparse.ArgumentParser(description="KVMD OTG MSD Helper")
parser.add_argument("--reset", action="store_true", help="Send SIGUSR1 to MSD kernel thread")
parser.add_argument("--set-image", dest="image_path", default=None, help="Change active image path")
parser.add_argument("--gadget", default="pikvm", help="USB gadget name")
options = parser.parse_args()
if options.reset:
_reset_msd()
if options.image_path is not None:
_set_msd_image(options.gadget, options.image_path)

24
kvmd/apps/otgmsd/__main__.py Normal file
View File

@ -0,0 +1,24 @@
# ========================================================================== #
# #
# 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 . import main
main()

4
setup.py
View File

@ -86,6 +86,8 @@ def main() -> None:
"kvmd.plugins.msd",
"kvmd.apps",
"kvmd.apps.kvmd",
"kvmd.apps.otg",
"kvmd.apps.otgmsd",
"kvmd.apps.htpasswd",
"kvmd.apps.cleanup",
"kvmd.apps.ipmi",
@ -104,6 +106,8 @@ def main() -> None:
entry_points={
"console_scripts": [
"kvmd = kvmd.apps.kvmd:main",
"kvmd-otg = kvmd.apps.otg:main",
"kvmd-otg-msd = kvmd.apps.otgmsd:main",
"kvmd-htpasswd = kvmd.apps.htpasswd:main",
"kvmd-cleanup = kvmd.apps.cleanup:main",
"kvmd-ipmi = kvmd.apps.ipmi:main",

1
testenv/Dockerfile
View File

@ -18,6 +18,7 @@ RUN pacman -Syu --noconfirm \
python-tox \
python-systemd \
python-dbus \
python-psutil \
python-mako \
nginx-mainline \
socat \