mirror of
https://github.com/openhop-dev/openhop_repeater.git
synced 2026-10-23 21:43:02 +00:00
agessaman
5bbc01f1c9
A bare number 0-255 named the first hash byte before multi-byte modes and still does. acl_clients compared its identity_hash filter against the first key byte only; it now matches the same prefix as remove.
882 lines
35 KiB
Python
882 lines
35 KiB
Python
"""ACL management over the web API: listing, setting, removing, and identity changes."""
|
|
|
|
import copy
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import cherrypy
|
|
import pytest
|
|
from openhop_core import LocalIdentity
|
|
from openhop_core.protocol import Identity
|
|
|
|
from repeater.data_acquisition.sqlite_handler import SQLiteHandler
|
|
from repeater.handler_helpers.login import LoginHelper
|
|
from repeater.handler_helpers.mesh_cli import MeshCLI
|
|
from repeater.web.api_endpoints import APIEndpoints
|
|
|
|
ROOM_SETTINGS = {"admin_password": "roomadmin", "guest_password": "roomguest"}
|
|
|
|
|
|
@pytest.fixture
|
|
def request_ctx(monkeypatch):
|
|
request = SimpleNamespace(method="GET", params={}, json={})
|
|
response = SimpleNamespace(headers={}, status=200)
|
|
monkeypatch.setattr(cherrypy, "request", request, raising=False)
|
|
monkeypatch.setattr(cherrypy, "response", response, raising=False)
|
|
return request
|
|
|
|
|
|
class _Daemon:
|
|
"""The parts of the daemon the ACL endpoints read, backed by a real store."""
|
|
|
|
def __init__(self, db, rooms):
|
|
self.local_identity = LocalIdentity()
|
|
self.login_helper = LoginHelper(
|
|
identity_manager=MagicMock(), packet_injector=AsyncMock(), sqlite_handler=db
|
|
)
|
|
self.login_helper.register_identity(
|
|
"repeater",
|
|
self.local_identity,
|
|
identity_type="repeater",
|
|
config={"repeater": {"security": {"admin_password": "adminpw"}}},
|
|
)
|
|
self.rooms = []
|
|
for name, identity in rooms:
|
|
self.add_room(name, identity)
|
|
self.identity_manager = SimpleNamespace(get_identities_by_type=self._by_type)
|
|
self.repeater_handler = SimpleNamespace(storage=SimpleNamespace(sqlite_handler=db))
|
|
|
|
def add_room(self, name, identity):
|
|
cfg = {"name": name, "type": "room_server", "settings": dict(ROOM_SETTINGS)}
|
|
self.login_helper.register_identity(name, identity, identity_type="room_server", config=cfg)
|
|
self.rooms.append((name, identity, cfg))
|
|
|
|
def _by_type(self, kind):
|
|
return list(self.rooms) if kind == "room_server" else []
|
|
|
|
|
|
def _api(daemon, config=None):
|
|
api = APIEndpoints.__new__(APIEndpoints)
|
|
api.config = config or {}
|
|
api.daemon_instance = daemon
|
|
api.config_manager = MagicMock()
|
|
api.config_manager.save_to_file.return_value = True
|
|
return api
|
|
|
|
|
|
def _post(request, api_method, body):
|
|
request.method = "POST"
|
|
request.json = body
|
|
return api_method()
|
|
|
|
|
|
@pytest.fixture
|
|
def db(tmp_path):
|
|
return SQLiteHandler(tmp_path)
|
|
|
|
|
|
def test_set_list_and_remove_an_entry(db, request_ctx):
|
|
daemon = _Daemon(db, [])
|
|
api = _api(daemon)
|
|
admin = LocalIdentity().get_public_key().hex()
|
|
|
|
result = _post(
|
|
request_ctx,
|
|
api.acl_set_permissions,
|
|
{"identity_name": "repeater", "client_pubkey": admin, "permissions": 3},
|
|
)
|
|
assert result["success"] is True
|
|
assert result["data"]["persisted"] is True
|
|
assert result["data"]["permissions"] == "admin"
|
|
|
|
request_ctx.method = "GET"
|
|
listed = api.acl_clients(identity_name="repeater")["data"]["clients"]
|
|
assert [(c["public_key_full"], c["permissions_value"], c["persisted"]) for c in listed] == [
|
|
(admin, 3, True)
|
|
]
|
|
assert listed[0]["identity_pubkey"] == daemon.local_identity.get_public_key().hex()
|
|
info = api.acl_info()["data"]["acls"][0]
|
|
# Provisioned, not logged in: an entry, stored, but not a session.
|
|
assert (info["acl_entries"], info["stored_entries"], info["authenticated_clients"]) == (1, 1, 0)
|
|
|
|
removed = _post(
|
|
request_ctx,
|
|
api.acl_remove_client,
|
|
{"client_pubkey": admin, "identity_name": "repeater"},
|
|
)
|
|
assert removed["success"] is True
|
|
assert removed["data"]["removed_from"] == ["repeater"]
|
|
assert db.load_acl_entries(daemon.local_identity.get_public_key().hex()) == []
|
|
|
|
|
|
def test_remove_accepts_the_legacy_public_key_field(db, request_ctx):
|
|
daemon = _Daemon(db, [])
|
|
api = _api(daemon)
|
|
admin = LocalIdentity().get_public_key()
|
|
daemon.login_helper.get_acl_by_name("repeater").apply_permissions(admin, 3)
|
|
|
|
removed = _post(request_ctx, api.acl_remove_client, {"public_key": admin.hex()})
|
|
assert removed["success"] is True
|
|
|
|
|
|
def test_remove_accepts_a_numeric_identity_hash(db, request_ctx):
|
|
daemon = _Daemon(db, [])
|
|
api = _api(daemon)
|
|
acl = daemon.login_helper.get_acl_by_name("repeater")
|
|
admin = LocalIdentity().get_public_key()
|
|
acl.apply_permissions(admin, 3)
|
|
hash_byte = acl.identity_pubkey_hex[:2]
|
|
|
|
body = {"identity_hash": int(hash_byte, 16), "client_pubkey": admin.hex()}
|
|
assert _post(request_ctx, api.acl_remove_client, body)["success"] is True
|
|
|
|
|
|
@pytest.mark.parametrize("path_hash_mode", [1, 2])
|
|
def test_remove_matches_the_multi_byte_hash_acl_clients_lists(db, request_ctx, path_hash_mode):
|
|
"""acl_clients lists a 2- or 3-byte identity_hash; remove must accept it back."""
|
|
daemon = _Daemon(db, [])
|
|
api = _api(daemon, {"mesh": {"path_hash_mode": path_hash_mode}})
|
|
acl = daemon.login_helper.get_acl_by_name("repeater")
|
|
admin = LocalIdentity().get_public_key()
|
|
acl.apply_permissions(admin, 3)
|
|
|
|
request_ctx.method = "GET"
|
|
listed = api.acl_clients(identity_name="repeater")["data"]["clients"]
|
|
identity_hash = listed[0]["identity_hash"]
|
|
assert len(identity_hash) == 2 + 2 * (path_hash_mode + 1)
|
|
|
|
removed = _post(
|
|
request_ctx,
|
|
api.acl_remove_client,
|
|
{"identity_hash": identity_hash, "client_pubkey": admin.hex()},
|
|
)
|
|
assert removed["success"] is True
|
|
assert removed["data"]["removed_from"] == ["repeater"]
|
|
assert db.load_acl_entries(daemon.local_identity.get_public_key().hex()) == []
|
|
|
|
|
|
def test_a_multi_byte_hash_does_not_select_an_identity_sharing_only_its_first_byte(db):
|
|
api = _api(_Daemon(db, []), {"mesh": {"path_hash_mode": 1}})
|
|
(owner,) = api._acl_owners()
|
|
key = owner[2].get_public_key()
|
|
other = bytes([key[0], key[1] ^ 0xFF])
|
|
assert api._acl_targets(None, f"0x{key[:2].hex().upper()}") == [owner]
|
|
assert api._acl_targets(None, f"0x{other.hex()}") == []
|
|
assert api._acl_targets(None, f"0X{key[:2].hex()}") == [owner]
|
|
# A number has no width of its own: 0-255 still names the first byte alone.
|
|
assert api._acl_targets(None, key[0]) == [owner]
|
|
assert api._acl_targets(None, str(key[0])) == [owner]
|
|
if key[0]: # a zero first byte drops out of a number
|
|
assert api._acl_targets(None, int.from_bytes(key[:2], "big")) == [owner]
|
|
|
|
|
|
def test_a_leading_zero_hash_byte_keeps_its_width(db):
|
|
api = _api(_Daemon(db, []), {"mesh": {"path_hash_mode": 1}})
|
|
(owner,) = api._acl_owners()
|
|
key = owner[2].get_public_key()
|
|
assert api._parse_hash_prefix("0x0042") == b"\x00\x42"
|
|
assert api._parse_hash_prefix("0x042") == b"\x00\x42"
|
|
assert api._parse_hash_prefix("0x42") == b"\x42"
|
|
assert api._acl_targets(None, f"0x00{key[1]:02x}") == ([owner] if key[0] == 0 else [])
|
|
|
|
|
|
def test_acl_clients_filters_on_the_multi_byte_hash(db, request_ctx):
|
|
daemon = _Daemon(db, [])
|
|
api = _api(daemon, {"mesh": {"path_hash_mode": 1}})
|
|
daemon.login_helper.get_acl_by_name("repeater").apply_permissions(
|
|
LocalIdentity().get_public_key(), 3
|
|
)
|
|
key = daemon.local_identity.get_public_key()
|
|
other = bytes([key[0], key[1] ^ 0xFF])
|
|
|
|
result = api.acl_clients(identity_hash=f"0x{key[:2].hex()}")
|
|
assert len(result["data"]["clients"]) == 1
|
|
assert api.acl_clients(identity_hash=f"0x{other.hex()}")["data"]["clients"] == []
|
|
|
|
|
|
@pytest.mark.parametrize("identity_hash", ["0x", "0xzz", "-1", "nope", 1.5, [1], True])
|
|
def test_remove_rejects_a_malformed_identity_hash(db, request_ctx, identity_hash):
|
|
api = _api(_Daemon(db, []))
|
|
body = {"identity_hash": identity_hash, "client_pubkey": "aa" * 32}
|
|
result = _post(request_ctx, api.acl_remove_client, body)
|
|
assert result["success"] is False
|
|
assert "identity_hash" in result["error"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"body, message",
|
|
[
|
|
({"client_pubkey": "aa" * 32, "permissions": 3}, "identity_name"),
|
|
({"identity_name": "repeater", "client_pubkey": "aa" * 16, "permissions": 3}, "64"),
|
|
({"identity_name": "repeater", "client_pubkey": "zz" * 32, "permissions": 3}, "hex"),
|
|
({"identity_name": "repeater", "client_pubkey": "aa" * 32, "permissions": "3"}, "integer"),
|
|
({"identity_name": "repeater", "client_pubkey": "aa" * 32, "permissions": True}, "integer"),
|
|
({"identity_name": "repeater", "client_pubkey": "aa" * 32, "permissions": 0}, "remove"),
|
|
({"identity_name": "repeater", "client_pubkey": "aa" * 32, "permissions": 128}, "remove"),
|
|
({"identity_name": "nope", "client_pubkey": "aa" * 32, "permissions": 3}, "not found"),
|
|
({"identity_name": "repeater", "client_pubkey": "ab" * 32, "permissions": 3}, "valid"),
|
|
],
|
|
)
|
|
def test_set_permissions_validates_its_input(db, request_ctx, body, message):
|
|
api = _api(_Daemon(db, []))
|
|
result = _post(request_ctx, api.acl_set_permissions, body)
|
|
assert result["success"] is False
|
|
assert message in result["error"]
|
|
|
|
|
|
def test_a_room_read_write_entry_is_reported_as_not_stored(db, request_ctx):
|
|
daemon = _Daemon(db, [("room-a", LocalIdentity())])
|
|
api = _api(daemon)
|
|
result = _post(
|
|
request_ctx,
|
|
api.acl_set_permissions,
|
|
{
|
|
"identity_name": "room-a",
|
|
"client_pubkey": LocalIdentity().get_public_key().hex(),
|
|
"permissions": 2,
|
|
},
|
|
)
|
|
assert result["success"] is True
|
|
assert result["data"]["persisted"] is False
|
|
assert "until restart" in result["message"]
|
|
|
|
|
|
def test_identities_that_share_a_hash_byte_are_listed_separately(db, request_ctx):
|
|
daemon = _Daemon(db, [])
|
|
twin = LocalIdentity()
|
|
while twin.get_public_key()[0] != daemon.local_identity.get_public_key()[0]:
|
|
twin = LocalIdentity()
|
|
daemon.add_room("room-a", twin)
|
|
api = _api(daemon)
|
|
for name in ("repeater", "room-a"):
|
|
_post(
|
|
request_ctx,
|
|
api.acl_set_permissions,
|
|
{
|
|
"identity_name": name,
|
|
"client_pubkey": LocalIdentity().get_public_key().hex(),
|
|
"permissions": 3,
|
|
},
|
|
)
|
|
|
|
request_ctx.method = "GET"
|
|
listed = api.acl_clients()["data"]["clients"]
|
|
assert sorted(c["identity_name"] for c in listed) == ["repeater", "room-a"]
|
|
|
|
|
|
def test_update_identity_moves_a_room_acl_when_renamed_and_rekeyed(db, request_ctx):
|
|
old_seed = "11" * 32
|
|
old_identity = LocalIdentity(seed=bytes.fromhex(old_seed))
|
|
daemon = _Daemon(db, [("room-a", old_identity)])
|
|
config = {
|
|
"identities": {
|
|
"room_servers": [
|
|
{"name": "room-a", "identity_key": old_seed, "settings": dict(ROOM_SETTINGS)}
|
|
]
|
|
}
|
|
}
|
|
api = _api(daemon, config)
|
|
admin = LocalIdentity()
|
|
daemon.login_helper.get_acl_by_name("room-a").apply_permissions(admin.get_public_key(), 3)
|
|
|
|
new_seed = "22" * 32
|
|
request_ctx.method = "PUT"
|
|
request_ctx.json = {"name": "room-a", "new_name": "room-b", "identity_key": new_seed}
|
|
assert api.update_identity()["success"] is True
|
|
|
|
new_identity = LocalIdentity(seed=bytes.fromhex(new_seed))
|
|
rows = db.load_acl_entries(new_identity.get_public_key().hex())
|
|
assert [r["client_pubkey"] for r in rows] == [admin.get_public_key().hex()]
|
|
assert db.load_acl_entries(old_identity.get_public_key().hex()) == []
|
|
|
|
|
|
def test_delete_identity_drops_the_room_acl(db, request_ctx):
|
|
seed = "33" * 32
|
|
identity = LocalIdentity(seed=bytes.fromhex(seed))
|
|
daemon = _Daemon(db, [("room-a", identity)])
|
|
daemon.identity_manager.named_identities = {}
|
|
config = {"identities": {"room_servers": [{"name": "room-a", "identity_key": seed}]}}
|
|
api = _api(daemon, config)
|
|
daemon.login_helper.get_acl_by_name("room-a").apply_permissions(
|
|
LocalIdentity().get_public_key(), 3
|
|
)
|
|
|
|
request_ctx.method = "DELETE"
|
|
assert api.delete_identity(name="room-a", type="room_server")["success"] is True
|
|
assert db.load_acl_entries(identity.get_public_key().hex()) == []
|
|
|
|
|
|
def test_web_cli_get_acl_is_local(db, request_ctx):
|
|
daemon = _Daemon(db, [])
|
|
acl = daemon.login_helper.get_acl_by_name("repeater")
|
|
admin = LocalIdentity().get_public_key()
|
|
acl.apply_permissions(admin, 3)
|
|
cli = MeshCLI(
|
|
"/tmp/cfg.yaml",
|
|
{"repeater": {}},
|
|
SimpleNamespace(save_to_file=MagicMock(return_value=True), live_update_daemon=MagicMock()),
|
|
acl=acl,
|
|
)
|
|
daemon.text_helper = SimpleNamespace(cli=cli)
|
|
api = _api(daemon)
|
|
|
|
# Past the auth decorator: authentication is not what this covers.
|
|
reply = _post(request_ctx, lambda: APIEndpoints.cli.__wrapped__(api), {"command": "get acl"})
|
|
assert reply["data"]["reply"] == "ACL:\n03 " + admin.hex().upper()
|
|
# The mesh path passes no local flag.
|
|
assert cli.handle_command(b"x", "get acl", is_admin=True).startswith("Error:")
|
|
|
|
|
|
def test_a_loaded_admin_has_a_secret_for_the_current_key(db):
|
|
# The UI's "stored" entries must be usable straight after a restart.
|
|
daemon = _Daemon(db, [])
|
|
admin = LocalIdentity()
|
|
daemon.login_helper.get_acl_by_name("repeater").apply_permissions(admin.get_public_key(), 3)
|
|
|
|
restarted = LoginHelper(
|
|
identity_manager=MagicMock(), packet_injector=AsyncMock(), sqlite_handler=db
|
|
)
|
|
restarted.register_identity(
|
|
"repeater", daemon.local_identity, identity_type="repeater", config={"repeater": {}}
|
|
)
|
|
client = restarted.get_acl_by_name("repeater").get_client(admin.get_public_key())
|
|
assert client.shared_secret == Identity(admin.get_public_key()).calc_shared_secret(
|
|
daemon.local_identity.get_private_key()
|
|
)
|
|
|
|
|
|
def _room_update_setup(db):
|
|
old_seed, other_seed = "11" * 32, "44" * 32
|
|
old_identity = LocalIdentity(seed=bytes.fromhex(old_seed))
|
|
daemon = _Daemon(db, [("room-a", old_identity)])
|
|
config = {
|
|
"identities": {
|
|
"room_servers": [
|
|
{"name": "room-a", "identity_key": old_seed, "settings": dict(ROOM_SETTINGS)},
|
|
{"name": "room-b", "identity_key": other_seed, "settings": dict(ROOM_SETTINGS)},
|
|
]
|
|
}
|
|
}
|
|
admin = LocalIdentity()
|
|
daemon.login_helper.get_acl_by_name("room-a").apply_permissions(admin.get_public_key(), 3)
|
|
return daemon, config, old_identity, admin
|
|
|
|
|
|
def test_update_identity_refuses_a_key_another_identity_uses(db, request_ctx):
|
|
daemon, config, old_identity, admin = _room_update_setup(db)
|
|
api = _api(daemon, config)
|
|
|
|
request_ctx.method = "PUT"
|
|
request_ctx.json = {"name": "room-a", "new_name": "renamed", "identity_key": "44" * 32}
|
|
result = api.update_identity()
|
|
|
|
assert result["success"] is False
|
|
assert "already used" in result["error"]
|
|
room = config["identities"]["room_servers"][0]
|
|
assert (room["name"], room["identity_key"]) == ("room-a", "11" * 32)
|
|
api.config_manager.save_to_file.assert_not_called()
|
|
assert len(db.load_acl_entries(old_identity.get_public_key().hex())) == 1
|
|
|
|
|
|
@pytest.mark.parametrize("taken", ["repeater", "comp-a"])
|
|
def test_a_room_cannot_take_a_name_another_identity_uses(db, request_ctx, taken):
|
|
"""Names are unique across every identity at boot, so a duplicate saved
|
|
here would leave a config the next restart refuses."""
|
|
daemon, config, old_identity, admin = _room_update_setup(db)
|
|
config["identities"]["companions"] = [{"name": "comp-a", "identity_key": "55" * 32}]
|
|
api = _api(daemon, config)
|
|
|
|
live_entry = config["identities"]["room_servers"][0]
|
|
|
|
request_ctx.method = "PUT"
|
|
request_ctx.json = {"name": "room-a", "new_name": taken}
|
|
result = api.update_identity()
|
|
|
|
assert result["success"] is False
|
|
# Restored in place: the identity manager holds this same dict.
|
|
assert config["identities"]["room_servers"][0] is live_entry
|
|
assert live_entry["name"] == "room-a"
|
|
api.config_manager.save_to_file.assert_not_called()
|
|
|
|
|
|
def test_a_new_room_cannot_reuse_another_identitys_key(db, request_ctx):
|
|
daemon, config, old_identity, admin = _room_update_setup(db)
|
|
api = _api(daemon, config)
|
|
|
|
request_ctx.method = "POST"
|
|
request_ctx.json = {
|
|
"name": "room-c",
|
|
"type": "room_server",
|
|
"identity_key": "44" * 32, # room-b's
|
|
"settings": dict(ROOM_SETTINGS),
|
|
}
|
|
result = api.create_identity()
|
|
|
|
assert result["success"] is False
|
|
assert "already used" in result["error"]
|
|
assert [r["name"] for r in config["identities"]["room_servers"]] == ["room-a", "room-b"]
|
|
api.config_manager.save_to_file.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("taken", ["repeater", "room-a"])
|
|
def test_a_new_companion_cannot_take_a_name_another_identity_uses(db, request_ctx, taken):
|
|
daemon, config, old_identity, admin = _room_update_setup(db)
|
|
api = _api(daemon, config)
|
|
|
|
request_ctx.method = "POST"
|
|
request_ctx.json = {"name": taken, "type": "companion", "identity_key": "66" * 32}
|
|
result = api.create_identity()
|
|
|
|
assert result["success"] is False
|
|
assert "already exists" in result["error"]
|
|
api.config_manager.save_to_file.assert_not_called()
|
|
|
|
|
|
def test_renaming_a_room_to_its_own_name_is_allowed(db, request_ctx):
|
|
daemon, config, old_identity, admin = _room_update_setup(db)
|
|
api = _api(daemon, config)
|
|
|
|
request_ctx.method = "PUT"
|
|
request_ctx.json = {"name": "room-a", "new_name": "room-a"}
|
|
assert api.update_identity()["success"] is True
|
|
|
|
|
|
@pytest.mark.parametrize("settings", ["x", {"flood_advert_interval_hours": "abc"}])
|
|
def test_bad_room_settings_undo_the_rename_in_the_same_request(db, request_ctx, settings):
|
|
daemon, config, old_identity, admin = _room_update_setup(db)
|
|
api = _api(daemon, config)
|
|
before = copy.deepcopy(config["identities"]["room_servers"][0])
|
|
|
|
request_ctx.method = "PUT"
|
|
request_ctx.json = {"name": "room-a", "new_name": "renamed", "settings": settings}
|
|
assert api.update_identity()["success"] is False
|
|
assert config["identities"]["room_servers"][0] == before
|
|
|
|
|
|
def test_identity_hash_zero_names_hash_byte_zero_not_every_acl(db):
|
|
"""0 is falsy; it must not fall through to "no filter, every ACL"."""
|
|
api = _api(_Daemon(db, []))
|
|
owners = api._acl_owners()
|
|
assert api._acl_targets(None, 0) == [o for o in owners if o[2].get_public_key()[0] == 0]
|
|
assert api._acl_targets(None, 0) != owners or not owners
|
|
|
|
|
|
def test_the_identity_endpoints_stay_exposed():
|
|
from repeater.web.api_endpoints import APIEndpoints
|
|
|
|
for name in ("create_identity", "update_identity", "delete_identity"):
|
|
assert getattr(getattr(APIEndpoints, name), "exposed", False), name
|
|
|
|
|
|
def test_update_identity_does_not_save_a_key_whose_acl_did_not_move(db, request_ctx):
|
|
daemon, config, old_identity, admin = _room_update_setup(db)
|
|
api = _api(daemon, config)
|
|
daemon.login_helper.move_room_acl = MagicMock(side_effect=RuntimeError("disk full"))
|
|
|
|
request_ctx.method = "PUT"
|
|
request_ctx.json = {"name": "room-a", "identity_key": "22" * 32}
|
|
result = api.update_identity()
|
|
|
|
assert result["success"] is False
|
|
api.config_manager.save_to_file.assert_not_called()
|
|
assert config["identities"]["room_servers"][0]["identity_key"] == "11" * 32
|
|
|
|
|
|
def test_a_failed_config_save_moves_the_acl_back(db, request_ctx):
|
|
daemon, config, old_identity, admin = _room_update_setup(db)
|
|
api = _api(daemon, config)
|
|
api.config_manager.save_to_file.return_value = False
|
|
|
|
request_ctx.method = "PUT"
|
|
request_ctx.json = {"name": "room-a", "identity_key": "22" * 32}
|
|
assert api.update_identity()["success"] is False
|
|
|
|
assert len(db.load_acl_entries(old_identity.get_public_key().hex())) == 1
|
|
new_identity = LocalIdentity(seed=bytes.fromhex("22" * 32))
|
|
assert db.load_acl_entries(new_identity.get_public_key().hex()) == []
|
|
|
|
|
|
def test_acl_info_uses_the_right_acl_when_hashes_collide(db, request_ctx):
|
|
daemon = _Daemon(db, [])
|
|
twin = LocalIdentity()
|
|
while twin.get_public_key()[0] != daemon.local_identity.get_public_key()[0]:
|
|
twin = LocalIdentity()
|
|
daemon.add_room("room-a", twin)
|
|
daemon.login_helper.get_acl_by_name("room-a").apply_permissions(
|
|
LocalIdentity().get_public_key(), 3
|
|
)
|
|
api = _api(daemon)
|
|
|
|
acls = {a["name"]: a for a in api.acl_info()["data"]["acls"]}
|
|
assert acls["repeater"]["acl_entries"] == 0
|
|
assert acls["room-a"]["acl_entries"] == 1
|
|
assert acls["room-a"]["has_admin_password"] is True
|
|
assert acls["repeater"]["store_error"] is None
|
|
|
|
|
|
def test_a_refused_update_leaves_the_live_config_unchanged(db, request_ctx):
|
|
daemon, config, old_identity, admin = _room_update_setup(db)
|
|
api = _api(daemon, config)
|
|
before = dict(config["identities"]["room_servers"][0], settings=dict(ROOM_SETTINGS))
|
|
|
|
request_ctx.method = "PUT"
|
|
request_ctx.json = {
|
|
"name": "room-a",
|
|
"new_name": "renamed",
|
|
"settings": {"admin_password": "same", "guest_password": "same"},
|
|
}
|
|
assert api.update_identity()["success"] is False
|
|
assert config["identities"]["room_servers"][0] == before
|
|
|
|
|
|
def test_a_key_change_with_no_store_is_refused(request_ctx):
|
|
config = {
|
|
"identities": {
|
|
"room_servers": [{"name": "room-a", "identity_key": "11" * 32, "settings": {}}]
|
|
}
|
|
}
|
|
api = _api(SimpleNamespace(), config)
|
|
request_ctx.method = "PUT"
|
|
request_ctx.json = {"name": "room-a", "identity_key": "22" * 32}
|
|
result = api.update_identity()
|
|
assert result["success"] is False
|
|
assert "no ACL store" in result["error"]
|
|
api.config_manager.save_to_file.assert_not_called()
|
|
|
|
|
|
def test_a_room_without_an_acl_does_not_borrow_one_by_hash(db, request_ctx):
|
|
daemon = _Daemon(db, [])
|
|
twin = LocalIdentity()
|
|
while twin.get_public_key()[0] != daemon.local_identity.get_public_key()[0]:
|
|
twin = LocalIdentity()
|
|
# Registered with the identity manager but, having no passwords, not for logins.
|
|
daemon.rooms.append(("room-b", twin, {"name": "room-b", "settings": {}}))
|
|
api = _api(daemon)
|
|
|
|
names = [a["name"] for a in api.acl_info()["data"]["acls"]]
|
|
assert names == ["repeater"]
|
|
result = _post(
|
|
request_ctx,
|
|
api.acl_set_permissions,
|
|
{
|
|
"identity_name": "room-b",
|
|
"client_pubkey": LocalIdentity().get_public_key().hex(),
|
|
"permissions": 3,
|
|
},
|
|
)
|
|
assert result["success"] is False
|
|
|
|
|
|
def test_acl_clients_reports_an_unreadable_store(db, request_ctx):
|
|
daemon = _Daemon(db, [])
|
|
daemon.login_helper.get_acl_by_name("repeater").load_error = "database is locked"
|
|
api = _api(daemon)
|
|
data = api.acl_clients()["data"]
|
|
assert data["store_errors"] == {"repeater": "database is locked"}
|
|
|
|
|
|
def test_remove_from_every_acl_reports_which_failed(db, request_ctx):
|
|
from repeater.handler_helpers.acl import ACLStoreError
|
|
|
|
daemon = _Daemon(db, [("room-a", LocalIdentity())])
|
|
key = LocalIdentity().get_public_key()
|
|
for name in ("repeater", "room-a"):
|
|
daemon.login_helper.get_acl_by_name(name).apply_permissions(key, 3)
|
|
daemon.login_helper.get_acl_by_name("room-a").remove_client = MagicMock(
|
|
side_effect=ACLStoreError("disk full")
|
|
)
|
|
api = _api(daemon)
|
|
|
|
result = _post(request_ctx, api.acl_remove_client, {"client_pubkey": key.hex()})
|
|
assert result["success"] is False
|
|
assert "room-a (disk full)" in result["error"]
|
|
assert "removed from repeater" in result["error"]
|
|
|
|
|
|
class _LiveDaemon:
|
|
"""A daemon whose hot reload goes through the real IdentityManager, as main.py's does."""
|
|
|
|
def __init__(self, db, room_name, room_seed):
|
|
from repeater.identity_manager import IdentityManager
|
|
|
|
self.identity_manager = IdentityManager({})
|
|
self.local_identity = LocalIdentity()
|
|
self.login_helper = LoginHelper(
|
|
identity_manager=self.identity_manager, packet_injector=AsyncMock(), sqlite_handler=db
|
|
)
|
|
self.repeater_handler = SimpleNamespace(storage=SimpleNamespace(sqlite_handler=db))
|
|
self.login_helper.register_identity(
|
|
"repeater", self.local_identity, identity_type="repeater", config={"repeater": {}}
|
|
)
|
|
cfg = {"name": room_name, "identity_key": room_seed, "settings": dict(ROOM_SETTINGS)}
|
|
self._register_identity_everywhere(
|
|
room_name, LocalIdentity(seed=bytes.fromhex(room_seed)), cfg, "room_server"
|
|
)
|
|
self.config = {"identities": {"room_servers": [cfg]}}
|
|
|
|
def _register_identity_everywhere(
|
|
self, name, identity, config, identity_type, previous_name=None
|
|
):
|
|
if not self.identity_manager.register_identity(
|
|
name=name, identity=identity, config=config, identity_type=identity_type
|
|
):
|
|
return False
|
|
self.login_helper.register_identity(
|
|
name=name, identity=identity, identity_type=identity_type, config=config
|
|
)
|
|
return True
|
|
|
|
|
|
def test_a_rename_applies_live_and_the_room_stays_manageable(db, request_ctx):
|
|
daemon = _LiveDaemon(db, "room-a", "11" * 32)
|
|
api = _api(daemon, daemon.config)
|
|
acl = daemon.login_helper.get_acl_by_name("room-a")
|
|
acl.apply_permissions(LocalIdentity().get_public_key(), 3)
|
|
|
|
request_ctx.method = "PUT"
|
|
request_ctx.json = {"name": "room-a", "new_name": "room-b"}
|
|
result = api.update_identity()
|
|
assert "applied immediately" in result["message"]
|
|
|
|
assert daemon.login_helper.get_acl_by_name("room-b") is acl
|
|
assert daemon.login_helper.get_acl_by_name("room-a") is None
|
|
added = _post(
|
|
request_ctx,
|
|
api.acl_set_permissions,
|
|
{
|
|
"identity_name": "room-b",
|
|
"client_pubkey": LocalIdentity().get_public_key().hex(),
|
|
"permissions": 3,
|
|
},
|
|
)
|
|
assert added["success"] is True
|
|
|
|
|
|
def test_a_rename_then_a_rekey_leaves_one_live_acl(db, request_ctx):
|
|
daemon = _LiveDaemon(db, "room-a", "11" * 32)
|
|
api = _api(daemon, daemon.config)
|
|
daemon.login_helper.get_acl_by_name("room-a").apply_permissions(
|
|
LocalIdentity().get_public_key(), 3
|
|
)
|
|
|
|
request_ctx.method = "PUT"
|
|
request_ctx.json = {"name": "room-a", "new_name": "room-b"}
|
|
api.update_identity()
|
|
request_ctx.method = "PUT"
|
|
request_ctx.json = {"name": "room-b", "identity_key": "22" * 32}
|
|
assert "applied immediately" in api.update_identity()["message"]
|
|
|
|
attached = [a for a in set(daemon.login_helper.acls_by_name.values()) if not a.detached]
|
|
assert len(attached) == 2 # the repeater and the room
|
|
assert [n for n, *_ in daemon.identity_manager.get_identities_by_type("room_server")] == [
|
|
"room-b"
|
|
]
|
|
new_pubkey = LocalIdentity(seed=bytes.fromhex("22" * 32)).get_public_key().hex()
|
|
assert daemon.login_helper.get_acl_by_name("room-b").store_key == new_pubkey
|
|
assert len(db.load_acl_entries(new_pubkey)) == 1
|
|
|
|
|
|
def test_a_refused_hot_reload_keeps_the_room_registered(db, request_ctx):
|
|
daemon = _LiveDaemon(db, "room-a", "11" * 32)
|
|
api = _api(daemon, daemon.config)
|
|
daemon._register_identity_everywhere = MagicMock(return_value=False)
|
|
|
|
request_ctx.method = "PUT"
|
|
request_ctx.json = {"name": "room-a", "new_name": "room-b"}
|
|
assert "Restart required" in api.update_identity()["message"]
|
|
assert [n for n, *_ in daemon.identity_manager.get_identities_by_type("room_server")] == [
|
|
"room-a"
|
|
]
|
|
|
|
|
|
def test_a_hot_reload_that_raises_keeps_the_room_registered(db, request_ctx):
|
|
daemon = _LiveDaemon(db, "room-a", "11" * 32)
|
|
api = _api(daemon, daemon.config)
|
|
daemon._register_identity_everywhere = MagicMock(side_effect=RuntimeError("boom"))
|
|
|
|
request_ctx.method = "PUT"
|
|
request_ctx.json = {"name": "room-a", "new_name": "room-b"}
|
|
assert "Restart required" in api.update_identity()["message"]
|
|
assert [n for n, *_ in daemon.identity_manager.get_identities_by_type("room_server")] == [
|
|
"room-a"
|
|
]
|
|
|
|
|
|
def test_a_rename_that_clears_the_passwords_waits_for_a_restart(db, request_ctx):
|
|
daemon = _LiveDaemon(db, "room-a", "11" * 32)
|
|
api = _api(daemon, daemon.config)
|
|
|
|
request_ctx.method = "PUT"
|
|
request_ctx.json = {
|
|
"name": "room-a",
|
|
"new_name": "room-b",
|
|
"settings": {"admin_password": "", "guest_password": ""},
|
|
}
|
|
assert "Restart required" in api.update_identity()["message"]
|
|
# Nothing half-applied: the running room keeps its registration and ACL.
|
|
assert [n for n, *_ in daemon.identity_manager.get_identities_by_type("room_server")] == [
|
|
"room-a"
|
|
]
|
|
assert daemon.login_helper.get_acl_by_name("room-a") is not None
|
|
|
|
|
|
def test_delete_identity_frees_the_name_and_hash(db, request_ctx):
|
|
daemon = _LiveDaemon(db, "room-a", "11" * 32)
|
|
api = _api(daemon, daemon.config)
|
|
|
|
request_ctx.method = "DELETE"
|
|
assert api.delete_identity(name="room-a", type="room_server")["success"] is True
|
|
identity = LocalIdentity(seed=bytes.fromhex("11" * 32))
|
|
# Re-creating the room registers at once instead of conflicting until a restart.
|
|
assert daemon.identity_manager.registration_error("room-a", identity, "room_server") is None
|
|
|
|
|
|
def test_giving_a_passwordless_room_a_password_registers_it_live(db, request_ctx):
|
|
daemon = _LiveDaemon(db, "room-a", "11" * 32)
|
|
api = _api(daemon, daemon.config)
|
|
request_ctx.method = "PUT"
|
|
request_ctx.json = {"name": "room-a", "settings": {"admin_password": "", "guest_password": ""}}
|
|
api.update_identity()
|
|
|
|
request_ctx.method = "PUT"
|
|
request_ctx.json = {"name": "room-a", "settings": {"admin_password": "new-admin"}}
|
|
result = api.update_identity()
|
|
assert "no reload needed" not in result["message"]
|
|
assert daemon.login_helper.get_acl_by_name("room-a").admin_password == "new-admin"
|
|
|
|
|
|
def test_a_hot_reload_names_the_room_it_replaces(db, request_ctx):
|
|
# The text helper stops the replaced RoomServer's sync loop by this name.
|
|
daemon = _LiveDaemon(db, "room-a", "11" * 32)
|
|
api = _api(daemon, daemon.config)
|
|
real = daemon._register_identity_everywhere
|
|
daemon._register_identity_everywhere = MagicMock(side_effect=real)
|
|
|
|
request_ctx.method = "PUT"
|
|
request_ctx.json = {"name": "room-a", "new_name": "room-b"}
|
|
api.update_identity()
|
|
assert daemon._register_identity_everywhere.call_args.kwargs["previous_name"] == "room-a"
|
|
|
|
|
|
def test_a_rekey_stops_the_old_key_answering_logins(db, request_ctx):
|
|
daemon = _LiveDaemon(db, "room-a", "11" * 32)
|
|
old_hash = LocalIdentity(seed=bytes.fromhex("11" * 32)).get_public_key()[0]
|
|
new_seed = "22" * 32
|
|
while LocalIdentity(seed=bytes.fromhex(new_seed)).get_public_key()[0] == old_hash:
|
|
new_seed = new_seed[2:] + "33"
|
|
daemon._unregister_identity_everywhere = lambda identity: (
|
|
daemon.login_helper.unregister_identity(identity)
|
|
)
|
|
api = _api(daemon, daemon.config)
|
|
assert old_hash in daemon.login_helper.handlers
|
|
|
|
request_ctx.method = "PUT"
|
|
request_ctx.json = {"name": "room-a", "identity_key": new_seed}
|
|
assert "applied immediately" in api.update_identity()["message"]
|
|
|
|
assert old_hash not in daemon.login_helper.handlers
|
|
new_hash = LocalIdentity(seed=bytes.fromhex(new_seed)).get_public_key()[0]
|
|
assert new_hash in daemon.login_helper.handlers
|
|
|
|
|
|
def test_deleting_a_room_stops_it_answering_logins(db, request_ctx):
|
|
daemon = _LiveDaemon(db, "room-a", "11" * 32)
|
|
room_hash = LocalIdentity(seed=bytes.fromhex("11" * 32)).get_public_key()[0]
|
|
daemon._unregister_identity_everywhere = lambda identity: (
|
|
daemon.login_helper.unregister_identity(identity)
|
|
)
|
|
api = _api(daemon, daemon.config)
|
|
|
|
request_ctx.method = "DELETE"
|
|
result = api.delete_identity(name="room-a", type="room_server")
|
|
assert "deactivated immediately" in result["message"]
|
|
assert room_hash not in daemon.login_helper.handlers
|
|
assert daemon.login_helper.get_acl_by_name("room-a") is None
|
|
|
|
|
|
def test_a_new_room_does_not_inherit_a_deleted_rooms_leftover_rows(db, request_ctx):
|
|
daemon = _LiveDaemon(db, "room-a", "11" * 32)
|
|
api = _api(daemon, daemon.config)
|
|
identity = LocalIdentity(seed=bytes.fromhex("11" * 32))
|
|
daemon.login_helper.get_acl_by_name("room-a").apply_permissions(
|
|
LocalIdentity().get_public_key(), 3
|
|
)
|
|
# The delete's ACL cleanup fails, leaving the old admin stored.
|
|
real_delete = db.delete_acl_label
|
|
db.delete_acl_label = MagicMock(side_effect=RuntimeError("disk full"))
|
|
request_ctx.method = "DELETE"
|
|
result = api.delete_identity(name="room-a", type="room_server")
|
|
assert "could not be removed" in result["message"]
|
|
db.delete_acl_label = real_delete
|
|
|
|
request_ctx.method = "POST"
|
|
request_ctx.json = {
|
|
"name": "room-a",
|
|
"type": "room_server",
|
|
"identity_key": "11" * 32,
|
|
"settings": dict(ROOM_SETTINGS),
|
|
}
|
|
api.create_identity()
|
|
assert db.load_acl_entries(identity.get_public_key().hex()) == []
|
|
|
|
|
|
def test_a_rename_is_refused_when_the_new_names_leftovers_cannot_be_cleared(db, request_ctx):
|
|
daemon = _LiveDaemon(db, "room-a", "11" * 32)
|
|
api = _api(daemon, daemon.config)
|
|
db.delete_acl_label = MagicMock(side_effect=RuntimeError("disk full"))
|
|
|
|
request_ctx.method = "PUT"
|
|
request_ctx.json = {"name": "room-a", "new_name": "room-b"}
|
|
result = api.update_identity()
|
|
assert result["success"] is False
|
|
assert daemon.config["identities"]["room_servers"][0]["name"] == "room-a"
|
|
api.config_manager.save_to_file.assert_not_called()
|
|
|
|
|
|
def test_creating_a_room_is_refused_when_leftovers_under_its_name_cannot_be_cleared(
|
|
db, request_ctx
|
|
):
|
|
daemon = _LiveDaemon(db, "room-a", "11" * 32)
|
|
api = _api(daemon, daemon.config)
|
|
db.delete_acl_label = MagicMock(side_effect=RuntimeError("disk full"))
|
|
|
|
request_ctx.method = "POST"
|
|
request_ctx.json = {
|
|
"name": "room-new",
|
|
"type": "room_server",
|
|
"identity_key": "44" * 32,
|
|
"settings": dict(ROOM_SETTINGS),
|
|
}
|
|
result = api.create_identity()
|
|
assert result["success"] is False
|
|
assert [r["name"] for r in daemon.config["identities"]["room_servers"]] == ["room-a"]
|
|
|
|
|
|
def test_acl_clients_names_a_client_from_its_advert_or_a_companion_contact(db, request_ctx):
|
|
daemon = _Daemon(db, [])
|
|
heard, via_companion, unknown = (LocalIdentity().get_public_key() for _ in range(3))
|
|
acl = daemon.login_helper.get_acl_by_name("repeater")
|
|
for key in (heard, via_companion, unknown):
|
|
acl.apply_permissions(key, 3)
|
|
now = 1_790_000_000
|
|
with db._connect() as conn:
|
|
conn.execute(
|
|
"INSERT INTO adverts (timestamp, pubkey, node_name, is_repeater, contact_type, "
|
|
"first_seen, last_seen, is_new_neighbor) VALUES (?, ?, ?, 0, 'Chat Node', ?, ?, 0)",
|
|
(now, heard.hex(), "Heard Node", now, now),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO companion_contacts (companion_hash, pubkey, name, updated_at) "
|
|
"VALUES ('0xab', ?, 'Contact Name', ?)",
|
|
(via_companion, now),
|
|
)
|
|
api = _api(daemon)
|
|
|
|
clients = {c["public_key_full"]: c for c in api.acl_clients()["data"]["clients"]}
|
|
assert (clients[heard.hex()]["client_name"], clients[heard.hex()]["client_type"]) == (
|
|
"Heard Node",
|
|
"Chat Node",
|
|
)
|
|
assert clients[via_companion.hex()]["client_name"] == "Contact Name"
|
|
assert clients[unknown.hex()]["client_name"] is None
|