First release

.gitignore

@@ -0,0 +1,21 @@
+### Build artifacts ###
+*.deb
+
+### VisualStudioCode ###
+.vscode/*
+
+### direnv ###
+.direnv
+.envrc
+
+### coverage ###
+.coverage
+htmlcov
+coverage.xml
+.coveragerc
+
+### cache ###
+.pytest_cache
+__pycache__
+
+.config

Jenkinsfile

@@ -0,0 +1,5 @@
+buildDebSbuild defaultTargets: 'bullseye-armhf',
+               defaultRunPythonChecks: true,
+               defaultRunLintian: true,
+               defaultRunCoverage: true,
+               defaultCoverageMin: "50"

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2025 Wiren Board
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -1,2 +1 @@
 # wb-homeui-auth
-Authentication service for Wiren Board web interface 

debian/changelog

@@ -0,0 +1,6 @@
+wb-homeui-auth (1.0.0) stable; urgency=medium
+
+  * Release
+
+ -- Petr Krasnoshchekov <[email protected]>  Wed, 20 Nov 2024 10:58:15 +0500
+

debian/compat

@@ -0,0 +1 @@
+10

debian/control

@@ -0,0 +1,16 @@
+Source: wb-homeui-auth
+Maintainer: Wiren Board team <[email protected]>
+Section: python
+Priority: optional
+Build-Depends: dh-python, debhelper (>= 10), python3-all, python3-setuptools, python3-pytest, python3-bcrypt, python3-jwt
+Standards-Version: 4.5.1
+X-Python3-Version: >= 3.9
+Homepage: https://github.com/wirenboard/wb-homeui-auth
+
+Package: wb-homeui-auth
+Architecture: all
+Depends: ${python3:Depends}, ${misc:Depends}, python3-bcrypt, python3-jwt
+Recommends: wb-mqtt-homeui (>= 2.108.0)
+Description: Authentication and authorization service for Wiren Board web interface
+ This package provides HTTP web-service for authentication
+ and authorization with Wiren Board web interface.

debian/rules

@@ -0,0 +1,10 @@
+#!/usr/bin/make -f
+
+export DH_VERBOSE=1
+export PYBUILD_NAME=wb_homeui_auth
+
+%:
+	dh $@ --with python3 --buildsystem=pybuild
+
+override_dh_installinit:
+	dh_installinit --noscripts

debian/wb-homeui-auth.install

@@ -0,0 +1 @@
+wb-homeui-auth usr/bin

debian/wb-homeui-auth.service

@@ -0,0 +1,14 @@
+[Unit]
+Description=Authentication and authorization service for Wiren Board web interface
+After=nginx.service
+
+[Service]
+Type=simple
+User=root
+ExecStart=/usr/bin/wb-homeui-auth
+Restart=on-failure
+RestartSec=2
+RestartPreventExitStatus=2 3 4 5 6 7
+
+[Install]
+WantedBy=multi-user.target

setup.py

@@ -0,0 +1,27 @@
+#!/usr/bin/env python3
+
+import re
+
+from setuptools import setup
+
+
+def get_version():
+    with open("debian/changelog", "r", encoding="utf-8") as f:
+        return re.match(r"wb-homeui-auth \((?P<version>.*)\)", f.readline()).group("version")
+
+
+setup(
+    name="wb-homeui-auth",
+    version=get_version(),
+    author="Petr Krasnoshchekov",
+    author_email="[email protected]",
+    maintainer="Wiren Board Team",
+    maintainer_email="[email protected]",
+    description="Authentication and authorization service for Wiren Board web interface",
+    license="MIT",
+    url="https://github.com/wirenboard/wb-homeui-auth",
+    packages=[
+        "wb.homeui_auth",
+    ],
+    test_suite="tests",
+)

tests/main_test.py

@@ -0,0 +1,149 @@
+import json
+import unittest
+from http.server import BaseHTTPRequestHandler
+from unittest.mock import MagicMock
+
+from wb.homeui_auth.http_response import (
+    response_200,
+    response_400,
+    response_401,
+    response_403,
+    response_404,
+)
+from wb.homeui_auth.main import (
+    WebRequestHandlerContext,
+    check_auth_handler,
+    delete_user_handler,
+    get_users_handler,
+)
+from wb.homeui_auth.users_storage import User, UsersStorage, UserType
+
+
+class DeleteUserHandlerTest(unittest.TestCase):
+    def setUp(self):
+        self.request = MagicMock()
+        self.context = WebRequestHandlerContext(users_storage=MagicMock())
+
+    def test_not_admin(self):
+        self.context.user = User("1", "user1", "password1", UserType.USER)
+        response = delete_user_handler(self.request, self.context)
+        self.assertEqual(response, response_403())
+
+    def test_not_authorized(self):
+        response = delete_user_handler(self.request, self.context)
+        self.assertEqual(response, response_403())
+
+    def test_bad_url(self):
+        self.request.path = "/users/aaaa/bbbb"
+        self.context.user = User("1", "user1", "password1", UserType.ADMIN)
+        response = delete_user_handler(self.request, self.context)
+        self.assertEqual(response, response_404())
+
+    def test_not_found(self):
+        self.request.path = "/users/aaaa"
+        self.context.user = User("1", "user1", "password1", UserType.ADMIN)
+        self.context.users_storage.get_user_by_id.return_value = None
+        response = delete_user_handler(self.request, self.context)
+        self.assertEqual(response, response_404())
+
+    def test_delete_self(self):
+        self.request.path = "/users/1"
+        self.context.user = User("1", "user1", "password1", UserType.ADMIN)
+        self.context.users_storage.get_user_by_id.return_value = self.context.user
+        response = delete_user_handler(self.request, self.context)
+
+        self.context.users_storage.get_user_by_id.assert_called_once_with(self.context.user.user_id)
+        self.assertEqual(response, response_400("Can't delete yourself"))
+
+    def test_success(self):
+        self.request.path = "/users/123"
+        self.context.user = User("1", "user1", "password1", UserType.ADMIN)
+        user_id = "123"
+        user = MagicMock()
+        user.user_id = user_id
+        self.context.users_storage.get_user_by_id.return_value = user
+        response = delete_user_handler(self.request, self.context)
+
+        self.context.users_storage.get_user_by_id.assert_called_once_with(user_id)
+        self.context.users_storage.delete_user.assert_called_once_with(user_id)
+        self.assertEqual(response, response_200())
+
+
+class GetUsersHandlerTests(unittest.TestCase):
+    def test_admin(self):
+        users_storage = MagicMock(spec=UsersStorage)
+        user = User("1", "user1", "password1", UserType.ADMIN)
+        users = [
+            User("1", "user1", "password1", UserType.USER),
+            User("2", "user2", "password2", UserType.ADMIN),
+        ]
+        users_storage.get_users.return_value = users
+
+        response = get_users_handler(None, WebRequestHandlerContext(user, users_storage))
+
+        expected_body = [
+            {"id": "1", "login": "user1", "type": UserType.USER.value},
+            {"id": "2", "login": "user2", "type": UserType.ADMIN.value},
+        ]
+        self.assertEqual(
+            response, response_200([["Content-type", "application/json"]], json.dumps(expected_body))
+        )
+
+    def test_non_admin(self):
+        user = User("1", "user1", "password1", UserType.USER)
+        response = get_users_handler(None, WebRequestHandlerContext(user))
+        self.assertEqual(response, response_403())
+
+        response = get_users_handler(None, WebRequestHandlerContext())
+        self.assertEqual(response, response_403())
+
+
+class CheckAuthHandlerTests(unittest.TestCase):
+    def setUp(self):
+        self.request = MagicMock(spec=BaseHTTPRequestHandler)
+        self.context = WebRequestHandlerContext(users_storage=MagicMock())
+
+    def test_no_required_user_type_no_user_without_users(self):
+        self.context.users_storage.has_users.return_value = False
+        self.request.headers = {}
+        response = check_auth_handler(self.request, self.context)
+        self.assertEqual(response, response_401())
+
+    def test_no_required_user_type_no_user_with_users(self):
+        self.context.users_storage.has_users.return_value = True
+        self.request.headers = {}
+        response = check_auth_handler(self.request, self.context)
+        self.assertEqual(response, response_401())
+
+    def test_required_user_no_user_without_users(self):
+        self.context.users_storage.has_users.return_value = False
+        self.request.headers = {"Required-User-Type": "user"}
+        response = check_auth_handler(self.request, self.context)
+        self.assertEqual(response, response_200())
+
+    def test_required_user_no_user_with_users(self):
+        self.context.users_storage.has_users.return_value = True
+        self.request.headers = {"Required-User-Type": "user"}
+        response = check_auth_handler(self.request, self.context)
+        self.assertEqual(response, response_401())
+
+    def test_required_user_user_with_users(self):
+        self.context.users_storage.has_users.return_value = True
+        self.request.headers = {"Required-User-Type": "user"}
+        self.context.user = User("1", "user", "password", UserType.USER)
+        response = check_auth_handler(self.request, self.context)
+        self.assertEqual(response, response_200(headers=[["User-Type", self.context.user.type.value]]))
+
+    def test_required_admin_user_with_users(self):
+        self.context.users_storage.has_users.return_value = True
+        self.request.headers = {"Required-User-Type": "admin"}
+        self.context.user = User("1", "user", "password", UserType.USER)
+        response = check_auth_handler(self.request, self.context)
+        self.assertEqual(response, response_403())
+
+    def test_required_user_admin_with_users(self):
+        self.context.users_storage.has_users.return_value = True
+        self.request.headers = {"Required-User-Type": "user"}
+        self.context.user = User("1", "user", "password", UserType.ADMIN)
+        response = check_auth_handler(self.request, self.context)
+        self.assertEqual(response, response_200(headers=[["User-Type", self.context.user.type.value]]))

wb-homeui-auth

@@ -0,0 +1,12 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import sys
+
+from wb.homeui_auth import main
+
+if __name__ == "__main__":
+    try:
+        sys.exit(main.main())
+    except KeyboardInterrupt:
+        pass

wb/homeui_auth/__init__.py

@@ -0,0 +1,5 @@
+import logging
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.DEBUG)
+logger.addHandler(logging.NullHandler())

wb/homeui_auth/http_response.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+from dataclasses import dataclass
+
+
+@dataclass
+class HttpResponse:
+    status: int
+    headers: list = None
+    body: str = None
+
+
+def response_200(headers: list = None, body: str = None) -> HttpResponse:
+    return HttpResponse(200, headers, body)
+
+
+def response_400(msg: str) -> HttpResponse:
+    return HttpResponse(400, body=f"Bad Request: {msg}")
+
+
+def response_401() -> HttpResponse:
+    return HttpResponse(401)
+
+
+def response_403() -> HttpResponse:
+    return HttpResponse(403)
+
+
+def response_404() -> HttpResponse:
+    return HttpResponse(404)
+
+
+def response_500(error: str) -> HttpResponse:
+    return HttpResponse(500, body=error)

wb/homeui_auth/keys_storage.py

@@ -0,0 +1,32 @@
+#!/usr/bin/env python3
+
+import secrets
+import string
+
+
+class KeysStorage:
+    def __init__(self, db_connection) -> None:
+        self.db_connection = db_connection
+        cursor = db_connection.cursor()
+        cursor.execute("CREATE TABLE IF NOT EXISTS keys (key TEXT NOT NULL)")
+        db_connection.commit()
+
+    def make_key(self) -> str:
+        alphabet = string.ascii_letters + string.digits + string.punctuation
+        key = "".join(secrets.choice(alphabet) for _ in range(32))
+        return key
+
+    def store_key(self, key: str) -> None:
+        cursor = self.db_connection.cursor()
+        cursor.execute("INSERT INTO keys (key) VALUES (?)", (key,))
+        self.db_connection.commit()
+
+    def get_key(self) -> str:
+        cursor = self.db_connection.cursor()
+        cursor.execute("SELECT key FROM keys")
+        result = cursor.fetchone()
+        if result is not None:
+            return result[0]
+        key = self.make_key()
+        self.store_key(key)
+        return key