Release v0.6.0
Add database-backed user management and password reset tooling.\n\nCo-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
rabadjiyski committed
1 parent
5547e55149
commit
c4a368d9f7
18 files changed
+1484
-154
No files matched your search
@@ -0,0 +1,329 @@
|
||||
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()
|
||||
Reference in new issue
Block a user