import io import json import os import sqlite3 import tempfile import threading import unittest from http.server import HTTPServer import urllib.error import urllib.request from unittest.mock import patch from cryptography.fernet import Fernet import webssh_backend.database as database import webssh_backend.http_api as http_api import password_reset from webssh_backend.auth import ( generate_token, hash_password, verify_password, verify_token_http, ) from webssh_backend.database import CredentialsManager class PasswordHashTests(unittest.TestCase): def test_password_hash_uses_salt_and_verifies_password(self): first_hash = hash_password("a-long-password") second_hash = hash_password("a-long-password") self.assertNotEqual(first_hash, second_hash) self.assertTrue(verify_password("a-long-password", first_hash)) self.assertFalse(verify_password("wrong-password", first_hash)) self.assertFalse(verify_password("a-long-password", "invalid-hash")) def test_password_change_token_is_limited_to_password_change(self): token = generate_token("user", password_change_required=True) self.assertIsNone(verify_token_http(token)) self.assertEqual(verify_token_http(token, allow_password_change=True), "user") class UserDatabaseTests(unittest.TestCase): def setUp(self): self.temp_dir = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.temp_dir.name, "credentials.db") def create_manager(self, users): with patch.object(database, "LEGACY_USERS", users): return CredentialsManager(self.db_path, Fernet.generate_key().decode()) def test_legacy_users_are_imported_once_with_forced_password_change(self): manager = self.create_manager([ ("owner", "legacy-owner-password"), ("operator", "legacy-user-password"), ]) owner = manager.get_user("owner") operator = manager.get_user("operator") self.assertEqual(owner["role"], "admin") self.assertEqual(operator["role"], "user") self.assertTrue(owner["must_change_password"]) self.assertTrue(operator["must_change_password"]) self.assertTrue(verify_password("legacy-owner-password", owner["password_hash"])) with patch.object(database, "LEGACY_USERS", [("intruder", "another-password")]): CredentialsManager(self.db_path, Fernet.generate_key().decode()) self.assertIsNone(manager.get_user("intruder")) def test_existing_user_table_is_migrated_with_auth_version(self): conn = sqlite3.connect(self.db_path) conn.execute(''' CREATE TABLE users ( username TEXT PRIMARY KEY, password_hash TEXT NOT NULL, role TEXT NOT NULL, must_change_password INTEGER NOT NULL DEFAULT 0, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ) ''') conn.execute(''' CREATE TABLE schema_migrations ( migration_id TEXT PRIMARY KEY, applied_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ) ''') conn.execute( "INSERT INTO schema_migrations (migration_id) VALUES ('001_users_table')" ) conn.execute( "INSERT INTO schema_migrations (migration_id) VALUES ('002_legacy_users_import')" ) conn.execute( "INSERT INTO users (username, password_hash, role) VALUES (?, ?, ?)", ("existing", hash_password("existing-account-password"), "admin"), ) conn.commit() conn.close() manager = self.create_manager([]) user = manager.get_user("existing") self.assertEqual(user["auth_version"], 0) self.assertEqual(user["role"], "admin") conn = sqlite3.connect(self.db_path) migration = conn.execute( "SELECT 1 FROM schema_migrations WHERE migration_id = '003_user_auth_version'" ).fetchone() conn.close() self.assertIsNotNone(migration) def test_first_admin_creation_is_only_allowed_once(self): manager = self.create_manager([]) self.assertTrue(manager.create_first_admin("owner", hash_password("new-long-password"))) self.assertFalse(manager.create_first_admin("second", hash_password("second-long-password"))) self.assertEqual(manager.get_user("owner")["role"], "admin") self.assertIsNone(manager.get_user("second")) def test_password_change_clears_required_flag(self): manager = self.create_manager([("owner", "legacy-password")]) self.assertTrue(manager.update_user_password("owner", hash_password("replacement-password"))) self.assertFalse(manager.get_user("owner")["must_change_password"]) def test_reset_requires_password_change_and_revokes_existing_tokens(self): manager = self.create_manager([]) manager.create_first_admin("owner", hash_password("original-account-password")) self.start_api(manager) status, login = self.request_json("POST", "/api/login", { "username": "owner", "password": "original-account-password", }) self.assertEqual(status, 200) existing_token = login["token"] user = manager.get_user("owner") self.assertTrue(manager.reset_user_password( "owner", hash_password("temporary-account-password"), )) reset_user = manager.get_user("owner") self.assertEqual(reset_user["auth_version"], user["auth_version"] + 1) self.assertTrue(reset_user["must_change_password"]) status, _ = self.request_json("GET", "/api/config", authorization=existing_token) self.assertEqual(status, 401) status, reset_login = self.request_json("POST", "/api/login", { "username": "owner", "password": "temporary-account-password", }) self.assertEqual(status, 200) self.assertTrue(reset_login["password_change_required"]) def test_password_reset_cli_updates_user_without_echoing_secret(self): manager = self.create_manager([]) manager.create_first_admin("owner", hash_password("original-account-password")) fake_input = io.StringIO() fake_output = io.StringIO() fake_input.isatty = lambda: True fake_output.isatty = lambda: True with ( patch.object(password_reset, "creds_manager", manager), patch.object(password_reset.sys, "stdin", fake_input), patch.object(password_reset.sys, "stdout", fake_output), patch.object(password_reset.sys, "argv", ["password-reset", "owner"]), patch.object(password_reset.getpass, "getpass", side_effect=[ "temporary-account-password", "temporary-account-password", ]), ): result = password_reset.main() self.assertEqual(result, 0) reset_user = manager.get_user("owner") self.assertTrue(reset_user["must_change_password"]) self.assertTrue(verify_password("temporary-account-password", reset_user["password_hash"])) self.assertNotIn("temporary-account-password", fake_output.getvalue()) def test_initial_setup_creates_only_one_admin(self): manager = self.create_manager([]) self.start_api(manager) status, data = self.request_json("GET", "/api/setup-status") self.assertEqual(status, 200) self.assertTrue(data["needs_setup"]) status, _ = self.request_json("POST", "/api/setup", { "username": "first-admin", "password": "initial-password-long", }) self.assertEqual(status, 201) self.assertEqual(manager.get_user("first-admin")["role"], "admin") status, _ = self.request_json("POST", "/api/setup", { "username": "second-admin", "password": "another-password-long", }) self.assertEqual(status, 409) def test_admin_can_manage_users_and_regular_users_cannot(self): manager = self.create_manager([]) manager.create_first_admin("owner", hash_password("owner-account-password")) self.start_api(manager) status, owner_login = self.request_json("POST", "/api/login", { "username": "owner", "password": "owner-account-password", }) self.assertEqual(status, 200) owner_token = owner_login["token"] status, _ = self.request_json( "POST", "/api/users", { "username": "operator", "password": "temporary-user-password", "role": "user", }, authorization=owner_token, ) self.assertEqual(status, 201) self.assertTrue(manager.get_user("operator")["must_change_password"]) status, listed = self.request_json("GET", "/api/users", authorization=owner_token) self.assertEqual(status, 200) self.assertEqual({user["username"] for user in listed["users"]}, {"owner", "operator"}) self.assertNotIn("password_hash", listed["users"][0]) status, _ = self.request_json( "POST", "/api/users/delete", {"username": "owner"}, authorization=owner_token, ) self.assertEqual(status, 409) manager.update_user_password("operator", hash_password("operator-account-password")) status, operator_login = self.request_json("POST", "/api/login", { "username": "operator", "password": "operator-account-password", }) self.assertEqual(status, 200) status, _ = self.request_json( "GET", "/api/users", authorization=operator_login["token"] ) self.assertEqual(status, 403) status, _ = self.request_json( "POST", "/api/users/delete", {"username": "operator"}, authorization=owner_token, ) self.assertEqual(status, 200) self.assertIsNone(manager.get_user("operator")) def test_imported_user_is_limited_until_password_is_changed(self): manager = self.create_manager([("owner", "legacy-owner-password")]) self.start_api(manager) status, login_data = self.request_json("POST", "/api/login", { "username": "owner", "password": "legacy-owner-password", }) self.assertEqual(status, 200) self.assertTrue(login_data["password_change_required"]) pending_token = login_data["token"] status, _ = self.request_json( "GET", "/api/config", authorization=pending_token ) self.assertEqual(status, 401) status, changed = self.request_json( "POST", "/api/change-password", {"password": "replacement-owner-password"}, authorization=pending_token, ) self.assertEqual(status, 200) status, _ = self.request_json("POST", "/api/verify", {"token": pending_token}) self.assertEqual(status, 401) status, verified = self.request_json("POST", "/api/verify", {"token": changed["token"]}) self.assertEqual(status, 200) self.assertTrue(verified["valid"]) self.assertEqual(verified["role"], "admin") def start_api(self, manager): self.server = HTTPServer(("127.0.0.1", 0), http_api.LoginHandler) self.server_thread = threading.Thread(target=self.server.serve_forever, daemon=True) self.server_thread.start() self.manager_patch = patch.object(http_api, "creds_manager", manager) self.manager_patch.start() self.base_url = f"http://127.0.0.1:{self.server.server_port}" def request_json(self, method, path, body=None, authorization=None): headers = {"Content-Type": "application/json"} if authorization: headers["Authorization"] = f"Bearer {authorization}" request = urllib.request.Request( self.base_url + path, data=json.dumps(body).encode() if body is not None else None, headers=headers, method=method, ) try: with urllib.request.urlopen(request) as response: return response.status, json.loads(response.read()) except urllib.error.HTTPError as error: return error.code, json.loads(error.read()) def tearDown(self): if hasattr(self, "manager_patch"): self.manager_patch.stop() self.server.shutdown() self.server.server_close() self.server_thread.join(timeout=2) self.temp_dir.cleanup() super().tearDown() if __name__ == "__main__": unittest.main()