LihonT/tests/test_portal.py

"""The private /lihont2 portal: real requests against a server on a throw-away database.

The credentials are random per run and live only in this process. The pages are checked for
the content each role must (and must not) see, for escaping of user-controlled text, and for
icon ids that exist in the shared sprite.
"""

import gc
import http.cookiejar
import importlib.util
import re
import secrets
import sys
import tempfile
import threading
import unittest
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
SERVER = ROOT / "src" / "web" / "server.py"
SPRITE = ROOT / "src" / "web" / "assets" / "site" / "icons.svg"


def load_server():
    spec = importlib.util.spec_from_file_location("lihont_portal_server", SERVER)
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, *args, **kwargs):
        return None


class PortalTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.portal = load_server()
        cls.directory = tempfile.TemporaryDirectory(prefix="lihont-portal-test-", ignore_cleanup_errors=True)
        base = Path(cls.directory.name)
        cls.apk = base / "lihont2.apk"
        cls.apk.write_bytes(b"PK\x03\x04" + b"0" * 2048)
        cls.admin_user = "admin_" + secrets.token_hex(4)
        cls.admin_password = secrets.token_urlsafe(18)
        cls.portal.ADMIN_USERNAME = cls.admin_user
        cls.portal.ADMIN_PASSWORD = cls.admin_password
        db = str(base / "data.db")
        cls.portal.init_db(db)
        handler = cls.portal.LihonTHandler
        handler.db_path = db
        handler.apk_path = str(cls.apk)
        handler._cached_stat = (0.0, 0, "")

        class Server(cls.portal.socketserver.ThreadingMixIn, cls.portal.http.server.HTTPServer):
            daemon_threads = True

        cls.httpd = Server(("127.0.0.1", 0), handler)
        cls.base_url = "http://127.0.0.1:%d%s" % (cls.httpd.server_address[1], cls.portal.BASE_PATH)
        threading.Thread(target=cls.httpd.serve_forever, daemon=True).start()
        cls.sprite_ids = set(re.findall(r'<symbol id="([^"]+)"', SPRITE.read_text(encoding="utf-8")))

    @classmethod
    def tearDownClass(cls):
        cls.httpd.shutdown()
        cls.httpd.server_close()
        gc.collect()  # the portal leaves sqlite connections to the collector; Windows will not delete an open file
        cls.directory.cleanup()

    def session(self, follow=True):
        jar = http.cookiejar.CookieJar()
        handlers = [urllib.request.HTTPCookieProcessor(jar)]
        if not follow:
            handlers.append(NoRedirect())
        return urllib.request.build_opener(*handlers)

    def fetch(self, opener, path, data=None):
        body = urllib.parse.urlencode(data).encode() if data is not None else None
        try:
            with opener.open(self.base_url + path, body) as response:
                return response.status, response.read().decode("utf-8"), response
        except urllib.error.HTTPError as error:
            return error.code, error.read().decode("utf-8"), error

    def login(self, user, password):
        opener = self.session()
        status, page, _ = self.fetch(opener, "/login", {"username": user, "password": password})
        return opener, status, page

    def assert_icons_exist(self, page):
        used = set(re.findall(r'icons\.svg#([a-z0-9-]+)', page))
        self.assertTrue(used, "page uses no icons at all")
        self.assertEqual(used - self.sprite_ids, set())

    def test_login_page_is_material(self):
        status, page, _ = self.fetch(self.session(), "/")
        self.assertEqual(status, 200)
        self.assertIn('/assets/site/md3.css', page)
        self.assertIn('class="field"', page)
        self.assertIn('type="password"', page)
        self.assertNotIn("user-scalable=no", page)
        self.assert_icons_exist(page)

    def test_wrong_password_shows_error_banner(self):
        status, page, _ = self.fetch(self.session(), "/login", {"username": self.admin_user, "password": "wrong"})
        self.assertEqual(status, 401)
        self.assertIn("Неверный логин или пароль", page)
        self.assertIn("alert--error", page)

    def test_download_needs_a_session(self):
        opener = self.session(follow=False)
        status, _, response = self.fetch(opener, "/download")
        self.assertEqual(status, 303)
        self.assertEqual(response.headers["Location"], self.portal.BASE_PATH + "/")

    def test_admin_flow(self):
        opener, status, page = self.login(self.admin_user, self.admin_password)
        self.assertEqual(status, 200)
        self.assertIn("Панель управления", page)
        self.assertIn(self.admin_user, page)
        self.assertIn(hash_prefix(self.apk), page)
        self.assert_icons_exist(page)

        tester, tester_password = "tester_" + secrets.token_hex(3), secrets.token_urlsafe(12)
        status, page, _ = self.fetch(opener, "/admin/create-user", {"username": tester, "password": tester_password, "note": "QA"})
        self.assertEqual(status, 200)
        self.assertIn(tester, page)
        self.assertIn("alert--success", page)
        self.assertIn("data-confirm", page)
        self.assertNotIn("onsubmit", page)

        status, page, _ = self.fetch(opener, "/admin/create-user", {"username": tester, "password": "x"})
        self.assertEqual(status, 400)
        self.assertIn("уже существует", page)

        status, page, _ = self.fetch(opener, "/admin/create-user", {"username": "", "password": ""})
        self.assertEqual(status, 400)
        self.assertIn("alert--error", page)

        user_opener, status, user_page = self.login(tester, tester_password)
        self.assertEqual(status, 200)
        self.assertIn("Release v2.0 Native", user_page)
        self.assertIn("Скачать lihont2.apk", user_page)
        self.assertNotIn("Панель управления", user_page)
        self.assertIn("data-copy", user_page)
        self.assert_icons_exist(user_page)

        status, body, response = self.fetch(user_opener, "/download")
        self.assertEqual(status, 200)
        self.assertEqual(response.headers["Content-Disposition"], 'attachment; filename="lihont2.apk"')
        self.assertEqual(response.headers["Content-Length"], str(self.apk.stat().st_size))

        user_id = re.search(r'name="user_id" value="(\d+)"', page).group(1)
        status, page, _ = self.fetch(opener, "/admin/delete-user", {"user_id": user_id})
        self.assertEqual(status, 200)
        self.assertIn("Пользователь удален.", page)
        self.assertNotIn(tester, page)

    def test_admin_cannot_delete_self(self):
        opener, _, page = self.login(self.admin_user, self.admin_password)
        status, page, _ = self.fetch(opener, "/admin/delete-user", {"user_id": "1"})
        self.assertIn(self.admin_user, page)

    def test_user_text_is_escaped(self):
        opener, _, _ = self.login(self.admin_user, self.admin_password)
        nasty = "<img src=x onerror=alert(1)>'\""
        status, page, _ = self.fetch(opener, "/admin/create-user", {"username": nasty, "password": "x", "note": "<b>note</b>"})
        self.assertEqual(status, 200)
        self.assertNotIn("<img src=x", page)
        self.assertNotIn("<b>note</b>", page)
        self.assertIn("&lt;img src=x", page)
        user_id = re.search(r'name="user_id" value="(\d+)"', page).group(1)
        self.fetch(opener, "/admin/delete-user", {"user_id": user_id})

    def test_missing_page_is_styled(self):
        status, page, _ = self.fetch(self.session(), "/nope")
        self.assertEqual(status, 404)
        self.assertIn("Страница не найдена", page)
        self.assertIn("/assets/site/md3.css", page)
        self.assert_icons_exist(page)

    def test_pages_use_only_shared_assets(self):
        _, login_page, _ = self.fetch(self.session(), "/")
        self.assertNotIn("<style", login_page)
        for href in re.findall(r'(?:href|src)="(/[^"]+)"', login_page):
            if href.startswith(self.portal.BASE_PATH):
                continue
            self.assertTrue(href.startswith("/assets/site/") or href == "/logo_single_note.svg", href)


def hash_prefix(path):
    import hashlib
    return hashlib.sha256(path.read_bytes()).hexdigest()[:16]


if __name__ == "__main__":
    unittest.main()