TOTP implementation

11 files changed, 155 insertions, 1 deletions

diff --git a/PKGBUILD b/PKGBUILD
index 84f23869..4fc82e79 100644
--- a/PKGBUILD
+++ b/PKGBUILD
@@ -45,6 +45,8 @@ depends=(
 	"python-aiohttp>=3.7.4.post0-1.1"
 	python-aiofiles
 	python-passlib
+	python-pyotp
+	python-qrcode
 	python-periphery
 	python-pyserial
 	python-pyserial-asyncio
@@ -121,6 +123,7 @@ md5sums=(SKIP)
 backup=(
 	etc/kvmd/{override,logging,auth,meta}.yaml
 	etc/kvmd/{ht,ipmi,vnc}passwd
+	etc/kvmd/totp.secret
 	etc/kvmd/nginx/{kvmd.ctx-{http,server},certbot.ctx-server}.conf
 	etc/kvmd/nginx/listen-http{,s}.conf
 	etc/kvmd/nginx/loc-{login,nocache,proxy,websocket,nobuffering,bigpost}.conf
@@ -162,6 +165,7 @@ package_kvmd() {
 	find "$pkgdir" -name ".gitignore" -delete
 	find "$_cfg_default" -type f -exec chmod 444 '{}' \;
 	chmod 400 "$_cfg_default/kvmd"/*passwd
+	chmod 400 "$_cfg_default/kvmd"/*.secret
 	chmod 750 "$_cfg_default/os/sudoers"
 	chmod 400 "$_cfg_default/os/sudoers"/*
 
@@ -176,6 +180,7 @@ package_kvmd() {
 
 	install -Dm644 -t "$pkgdir/etc/kvmd" "$_cfg_default/kvmd"/*.yaml
 	install -Dm600 -t "$pkgdir/etc/kvmd" "$_cfg_default/kvmd"/*passwd
+	install -Dm600 -t "$pkgdir/etc/kvmd" "$_cfg_default/kvmd"/*.secret
 	install -Dm644 -t "$pkgdir/etc/kvmd" "$_cfg_default/kvmd"/web.css
 	mkdir -p "$pkgdir/etc/kvmd/override.d"
 
diff --git a/configs/kvmd/totpasswd b/configs/kvmd/totpasswd
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/configs/kvmd/totpasswd
diff --git a/kvmd.install b/kvmd.install
index 9a1ca0aa..20454dda 100644
--- a/kvmd.install
+++ b/kvmd.install
@@ -15,6 +15,7 @@ post_upgrade() {
 	done
 
 	chown kvmd:kvmd /etc/kvmd/htpasswd || true
+	chown kvmd:kvmd /etc/kvmd/totp.secret || true
 	chown kvmd-ipmi:kvmd-ipmi /etc/kvmd/ipmipasswd || true
 	chown kvmd-vnc:kvmd-vnc /etc/kvmd/vncpasswd || true
 	chmod 600 /etc/kvmd/*passwd || true
diff --git a/kvmd/apps/__init__.py b/kvmd/apps/__init__.py
index 7d786bc4..cbc2400e 100644
--- a/kvmd/apps/__init__.py
+++ b/kvmd/apps/__init__.py
@@ -365,6 +365,12 @@ def _get_config_scheme() -> dict:
                     "type": Option("", type=valid_stripped_string),
                     # Dynamic content
                 },
+
+                "totp": {
+                    "secret": {
+                        "file": Option("/etc/kvmd/totp.secret", type=valid_abs_path, if_empty=""),
+                    },
+                },
             },
 
             "info": {  # Accessed via global config, see kvmd/info for details
diff --git a/kvmd/apps/kvmd/__init__.py b/kvmd/apps/kvmd/__init__.py
index 4d891f40..a90cd30b 100644
--- a/kvmd/apps/kvmd/__init__.py
+++ b/kvmd/apps/kvmd/__init__.py
@@ -82,6 +82,8 @@ def main(argv: (list[str] | None)=None) -> None:
 
             external_type=config.auth.external.type,
             external_kwargs=(config.auth.external._unpack(ignore=["type"]) if config.auth.external.type else {}),
+
+            totp_secret_path=config.auth.totp.secret.file,
         ),
         info_manager=InfoManager(global_config),
         log_reader=(LogReader() if config.log_reader.enabled else None),
diff --git a/kvmd/apps/kvmd/auth.py b/kvmd/apps/kvmd/auth.py
index 6ae696ad..02639eb0 100644
--- a/kvmd/apps/kvmd/auth.py
+++ b/kvmd/apps/kvmd/auth.py
@@ -21,6 +21,7 @@
 
 
 import secrets
+import pyotp
 
 from ...logging import get_logger
 
@@ -42,6 +43,8 @@ class AuthManager:
 
         external_type: str,
         external_kwargs: dict,
+
+        totp_secret_path: str,
     ) -> None:
 
         self.__enabled = enabled
@@ -53,12 +56,14 @@ class AuthManager:
             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.__force_internal_users = force_internal_users
+
         self.__external_service: (BaseAuthService | None) = None
         if enabled 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())
 
-        self.__force_internal_users = force_internal_users
+        self.__totp_secret_path = totp_secret_path
 
         self.__tokens: dict[str, str] = {}  # {token: user}
 
@@ -71,6 +76,16 @@ class AuthManager:
         assert self.__enabled
         assert self.__internal_service
 
+        if self.__totp_secret_path:
+            with open(self.__totp_secret_path) as secret_file:
+                secret = secret_file.read().strip()
+            if secret:
+                code = passwd[-6:]
+                if not pyotp.TOTP(secret).verify(code):
+                    get_logger().error("Got access denied for user %r by TOTP", user)
+                    return False
+                passwd = passwd[:-6]
+
         if user not in self.__force_internal_users and self.__external_service:
             service = self.__external_service
         else:
diff --git a/kvmd/apps/totp/__init__.py b/kvmd/apps/totp/__init__.py
new file mode 100644
index 00000000..64be9b90
--- /dev/null
+++ b/kvmd/apps/totp/__init__.py
@@ -0,0 +1,93 @@
+# ========================================================================== #
+#                                                                            #
+#    KVMD - The main PiKVM daemon.                                           #
+#                                                                            #
+#    Copyright (C) 2018-2022  Maxim Devaev <[email protected]>               #
+#                                                                            #
+#    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 argparse
+
+import pyotp
+import qrcode
+
+from ...yamlconf import Section
+
+from .. import init
+
+
+# =====
+def _get_secret_path(config: Section) -> str:
+    path: str = config.kvmd.auth.totp.secret.file
+    if len(path) == 0:
+        raise SystemExit("Error: TOTP file path is empty (i.e. it was disabled)")
+    return path
+
+
+def _read_secret(config: Section) -> str:
+    with open(_get_secret_path(config)) as file:
+        return file.read().strip()
+
+
+# =====
+def _cmd_init(config: Section, options: argparse.Namespace) -> None:
+    if not options.force:
+        if _read_secret(config):
+            raise SystemExit("Error: the TOTP secret already exists")
+    with open(_get_secret_path(config), "w") as file:
+        file.write(pyotp.random_base32())
+    _cmd_show(config, options)
+
+
+def _cmd_show(config: Section, _: argparse.Namespace) -> None:
+    secret = _read_secret(config)
+    if len(secret) == 0:
+        raise SystemExit("Error: TOTP secret is not configured")
+    uri = pyotp.totp.TOTP(secret).provisioning_uri(issuer_name="PiKVM")
+    qr = qrcode.QRCode()
+    qr.add_data(uri)
+    print()
+    print(uri)
+    print()
+    qr.print_ascii(invert=True)
+    print()
+
+
+# =====
+def main(argv: (list[str] | None)=None) -> None:
+    (parent_parser, argv, config) = init(
+        add_help=False,
+        cli_logging=True,
+        argv=argv,
+    )
+    parser = argparse.ArgumentParser(
+        prog="kvmd-totp",
+        description="Manage KVMD TOTP secret",
+        parents=[parent_parser],
+    )
+    parser.set_defaults(cmd=(lambda *_: parser.print_help()))
+    subparsers = parser.add_subparsers()
+
+    cmd_setup_parser = subparsers.add_parser("init", help="Generate and show TOTP secret with QR code")
+    cmd_setup_parser.add_argument("-f", "--force", action="store_true", help="Overwrite an existing secret")
+    cmd_setup_parser.set_defaults(cmd=_cmd_init)
+
+    cmd_show_parser = subparsers.add_parser("show", help="Show the current TOTP secret with QR code")
+    cmd_show_parser.set_defaults(cmd=_cmd_show)
+
+    options = parser.parse_args(argv[1:])
+    options.cmd(config, options)
diff --git a/kvmd/apps/totp/__main__.py b/kvmd/apps/totp/__main__.py
new file mode 100644
index 00000000..3849d1b9
--- /dev/null
+++ b/kvmd/apps/totp/__main__.py
@@ -0,0 +1,24 @@
+# ========================================================================== #
+#                                                                            #
+#    KVMD - The main PiKVM daemon.                                           #
+#                                                                            #
+#    Copyright (C) 2018-2022  Maxim Devaev <[email protected]>               #
+#                                                                            #
+#    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()
diff --git a/setup.py b/setup.py
index 36bb2d63..c6f3bc40 100755
--- a/setup.py
+++ b/setup.py
@@ -102,6 +102,7 @@ def main() -> None:
             "kvmd.apps.otgmsd",
             "kvmd.apps.otgconf",
             "kvmd.apps.htpasswd",
+            "kvmd.apps.totp",
             "kvmd.apps.edidconf",
             "kvmd.apps.cleanup",
             "kvmd.apps.ipmi",
@@ -128,6 +129,7 @@ def main() -> None:
                 "kvmd-otgmsd = kvmd.apps.otgmsd:main",
                 "kvmd-otgconf = kvmd.apps.otgconf:main",
                 "kvmd-htpasswd = kvmd.apps.htpasswd:main",
+                "kvmd-totp = kvmd.apps.totp:main",
                 "kvmd-edidconf = kvmd.apps.edidconf:main",
                 "kvmd-cleanup = kvmd.apps.cleanup:main",
                 "kvmd-ipmi = kvmd.apps.ipmi:main",
diff --git a/testenv/Dockerfile b/testenv/Dockerfile
index 21c3dfa6..c9bb5628 100644
--- a/testenv/Dockerfile
+++ b/testenv/Dockerfile
@@ -43,6 +43,8 @@ RUN pacman --noconfirm --ask=4 -Syy \
 		python-aiofiles \
 		python-periphery \
 		python-passlib \
+		python-pyotp \
+		python-qrcode \
 		python-pyserial \
 		python-setproctitle \
 		python-psutil \
diff --git a/testenv/tests/apps/kvmd/test_auth.py b/testenv/tests/apps/kvmd/test_auth.py
index 8b8577bd..3e47a6f0 100644
--- a/testenv/tests/apps/kvmd/test_auth.py
+++ b/testenv/tests/apps/kvmd/test_auth.py
@@ -59,6 +59,8 @@ async def _get_configured_manager(
 
         external_type=("htpasswd" if external_path else ""),
         external_kwargs=(_make_service_kwargs(external_path) if external_path else {}),
+
+        totp_secret_path="",
     )
 
     try:
@@ -149,6 +151,8 @@ async def test_ok__disabled() -> None:
 
             external_type="",
             external_kwargs={},
+
+            totp_secret_path="",
         )
 
         assert not manager.is_auth_enabled()