mirror of
https://github.com/PR0M3TH3AN/SeedPass.git
synced 2026-10-27 07:52:03 +00:00
754 lines
22 KiB
Python
754 lines
22 KiB
Python
from types import SimpleNamespace
|
|
from pathlib import Path
|
|
import os
|
|
import base64
|
|
import pytest
|
|
|
|
from seedpass import api
|
|
import string
|
|
from seedpass.core.password_generation import PasswordGenerator, PasswordPolicy
|
|
from seedpass.core.encryption import EncryptionManager
|
|
from nostr.client import NostrClient, DEFAULT_RELAYS
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_and_modify_totp_entry(client):
|
|
cl, token = client
|
|
calls = {}
|
|
|
|
def add_totp(label, seed, **kwargs):
|
|
calls["create"] = kwargs
|
|
return "uri"
|
|
|
|
def modify(idx, **kwargs):
|
|
calls["modify"] = (idx, kwargs)
|
|
|
|
api.app.state.pm.entry_manager.add_totp = add_totp
|
|
api.app.state.pm.entry_manager.modify_entry = modify
|
|
api.app.state.pm.entry_manager.get_next_index = lambda: 5
|
|
api.app.state.pm.parent_seed = "seed"
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
res = await cl.post(
|
|
"/api/v1/entry",
|
|
json={
|
|
"type": "totp",
|
|
"label": "T",
|
|
"index": 1,
|
|
"secret": "abc",
|
|
"period": 60,
|
|
"digits": 8,
|
|
"notes": "n",
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
assert res.json() == {"id": 5, "uri": "uri"}
|
|
assert calls["create"] == {
|
|
"index": 1,
|
|
"secret": "abc",
|
|
"period": 60,
|
|
"digits": 8,
|
|
"notes": "n",
|
|
"archived": False,
|
|
"deterministic": False,
|
|
}
|
|
|
|
res = await cl.put(
|
|
"/api/v1/entry/5",
|
|
json={"period": 90, "digits": 6},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
assert calls["modify"][0] == 5
|
|
assert calls["modify"][1]["period"] == 90
|
|
assert calls["modify"][1]["digits"] == 6
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_and_modify_ssh_entry(client):
|
|
cl, token = client
|
|
calls = {}
|
|
|
|
def add_ssh(label, seed, **kwargs):
|
|
calls["create"] = kwargs
|
|
return 2
|
|
|
|
def modify(idx, **kwargs):
|
|
calls["modify"] = (idx, kwargs)
|
|
|
|
api.app.state.pm.entry_manager.add_ssh_key = add_ssh
|
|
api.app.state.pm.entry_manager.modify_entry = modify
|
|
api.app.state.pm.parent_seed = "seed"
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
res = await cl.post(
|
|
"/api/v1/entry",
|
|
json={"type": "ssh", "label": "S", "index": 2, "notes": "n"},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
assert res.json() == {"id": 2}
|
|
assert calls["create"] == {"index": 2, "notes": "n", "archived": False}
|
|
|
|
res = await cl.put(
|
|
"/api/v1/entry/2",
|
|
json={"notes": "x"},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
assert calls["modify"][0] == 2
|
|
assert calls["modify"][1]["notes"] == "x"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_update_entry_error(client):
|
|
cl, token = client
|
|
|
|
def modify(*a, **k):
|
|
raise ValueError("nope")
|
|
|
|
api.app.state.pm.entry_manager.modify_entry = modify
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
res = await cl.put("/api/v1/entry/1", json={"username": "x"}, headers=headers)
|
|
assert res.status_code == 400
|
|
assert res.json() == {"detail": "nope"}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_update_config_secret_mode(client):
|
|
cl, token = client
|
|
called = {}
|
|
|
|
def set_secret(val):
|
|
called["val"] = val
|
|
|
|
api.app.state.pm.config_manager.set_secret_mode_enabled = set_secret
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
res = await cl.put(
|
|
"/api/v1/config/secret_mode_enabled",
|
|
json={"value": True},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
assert res.json() == {"status": "ok"}
|
|
assert called["val"] is True
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_totp_export_endpoint(client):
|
|
cl, token = client
|
|
api.app.state.pm.entry_manager.export_totp_entries = lambda seed: {"entries": ["x"]}
|
|
api.app.state.pm.parent_seed = "seed"
|
|
headers = {"Authorization": f"Bearer {token}", "X-SeedPass-Password": "pw"}
|
|
res = await cl.get("/api/v1/totp/export", headers=headers)
|
|
assert res.status_code == 200
|
|
assert res.json() == {"entries": ["x"]}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_totp_codes_endpoint(client):
|
|
cl, token = client
|
|
api.app.state.pm.entry_manager.list_entries = lambda **kw: [
|
|
(0, "Email", None, None, False)
|
|
]
|
|
api.app.state.pm.entry_manager.get_totp_code = lambda i, s: "123456"
|
|
api.app.state.pm.entry_manager.get_totp_time_remaining = lambda i: 30
|
|
api.app.state.pm.parent_seed = "seed"
|
|
headers = {"Authorization": f"Bearer {token}", "X-SeedPass-Password": "pw"}
|
|
res = await cl.get("/api/v1/totp", headers=headers)
|
|
assert res.status_code == 200
|
|
assert res.json() == {
|
|
"codes": [
|
|
{"id": 0, "label": "Email", "code": "123456", "seconds_remaining": 30}
|
|
]
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_parent_seed_endpoint_removed(client):
|
|
cl, token = client
|
|
res = await cl.get(
|
|
"/api/v1/parent-seed", headers={"Authorization": f"Bearer {token}"}
|
|
)
|
|
assert res.status_code == 404
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_fingerprint_endpoints(client):
|
|
cl, token = client
|
|
calls = {}
|
|
|
|
api.app.state.pm.add_new_fingerprint = lambda: calls.setdefault("add", True)
|
|
api.app.state.pm.fingerprint_manager.remove_fingerprint = (
|
|
lambda fp: calls.setdefault("remove", fp)
|
|
)
|
|
api.app.state.pm.select_fingerprint = lambda fp: calls.setdefault("select", fp)
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
res = await cl.post("/api/v1/fingerprint", headers=headers)
|
|
assert res.status_code == 200
|
|
assert res.json() == {"status": "ok"}
|
|
assert calls.get("add") is True
|
|
|
|
res = await cl.delete("/api/v1/fingerprint/abc", headers=headers)
|
|
assert res.status_code == 200
|
|
assert res.json() == {"status": "deleted"}
|
|
assert calls.get("remove") == "abc"
|
|
|
|
res = await cl.post(
|
|
"/api/v1/fingerprint/select",
|
|
json={"fingerprint": "xyz"},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
assert res.json() == {"status": "ok"}
|
|
assert calls.get("select") == "xyz"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_checksum_endpoints(client):
|
|
cl, token = client
|
|
calls = {}
|
|
|
|
api.app.state.pm.handle_verify_checksum = lambda: calls.setdefault("verify", True)
|
|
api.app.state.pm.handle_update_script_checksum = lambda: calls.setdefault(
|
|
"update", True
|
|
)
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
res = await cl.post("/api/v1/checksum/verify", headers=headers)
|
|
assert res.status_code == 200
|
|
assert res.json() == {"status": "ok"}
|
|
assert calls.get("verify") is True
|
|
|
|
res = await cl.post("/api/v1/checksum/update", headers=headers)
|
|
assert res.status_code == 200
|
|
assert res.json() == {"status": "ok"}
|
|
assert calls.get("update") is True
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_vault_import_via_path(client, tmp_path):
|
|
cl, token = client
|
|
called = {}
|
|
|
|
def import_db(path):
|
|
called["path"] = path
|
|
|
|
api.app.state.pm.handle_import_database = import_db
|
|
api.app.state.pm.sync_vault = lambda: called.setdefault("sync", True)
|
|
api.app.state.pm.encryption_manager = SimpleNamespace(
|
|
resolve_relative_path=lambda p: p
|
|
)
|
|
file_path = tmp_path / "b.json.enc"
|
|
file_path.write_text("{}")
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
res = await cl.post(
|
|
"/api/v1/vault/import",
|
|
json={"path": str(file_path)},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
assert res.json() == {"status": "ok"}
|
|
assert called["path"] == file_path
|
|
assert called.get("sync") is True
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_vault_import_via_upload(client, tmp_path):
|
|
cl, token = client
|
|
called = {}
|
|
|
|
def import_db(path):
|
|
called["path"] = path
|
|
|
|
api.app.state.pm.handle_import_database = import_db
|
|
api.app.state.pm.sync_vault = lambda: called.setdefault("sync", True)
|
|
file_path = tmp_path / "c.json"
|
|
file_path.write_text("{}")
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
with open(file_path, "rb") as fh:
|
|
res = await cl.post(
|
|
"/api/v1/vault/import",
|
|
files={"file": ("c.json", fh.read())},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
assert res.json() == {"status": "ok"}
|
|
assert isinstance(called.get("path"), Path)
|
|
assert called.get("sync") is True
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_vault_import_upload_sets_secure_temp_permissions(client, monkeypatch):
|
|
cl, token = client
|
|
called = {}
|
|
chmod_calls = []
|
|
unlink_calls = []
|
|
|
|
def import_db(path):
|
|
called["path"] = path
|
|
|
|
api.app.state.pm.handle_import_database = import_db
|
|
api.app.state.pm.sync_vault = lambda: called.setdefault("sync", True)
|
|
|
|
real_chmod = api.os.chmod
|
|
real_unlink = api.os.unlink
|
|
|
|
def tracking_chmod(path, mode):
|
|
chmod_calls.append((Path(path), mode))
|
|
return real_chmod(path, mode)
|
|
|
|
def tracking_unlink(path):
|
|
unlink_calls.append(Path(path))
|
|
return real_unlink(path)
|
|
|
|
monkeypatch.setattr(api.os, "chmod", tracking_chmod)
|
|
monkeypatch.setattr(api.os, "unlink", tracking_unlink)
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
res = await cl.post(
|
|
"/api/v1/vault/import",
|
|
files={"file": ("c.json.enc", b"{}")},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
assert res.json() == {"status": "ok"}
|
|
tmp_path_used = called["path"]
|
|
assert any(path == tmp_path_used and mode == 0o600 for path, mode in chmod_calls)
|
|
assert tmp_path_used in unlink_calls
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_vault_import_upload_temp_file_removed_on_failure(client, monkeypatch):
|
|
cl, token = client
|
|
called = {}
|
|
unlink_calls = []
|
|
|
|
def import_db(path):
|
|
called["path"] = path
|
|
raise ValueError("boom")
|
|
|
|
api.app.state.pm.handle_import_database = import_db
|
|
api.app.state.pm.sync_vault = lambda: called.setdefault("sync", True)
|
|
|
|
real_unlink = api.os.unlink
|
|
|
|
def tracking_unlink(path):
|
|
unlink_calls.append(Path(path))
|
|
return real_unlink(path)
|
|
|
|
monkeypatch.setattr(api.os, "unlink", tracking_unlink)
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
with pytest.raises(ValueError):
|
|
await cl.post(
|
|
"/api/v1/vault/import",
|
|
files={"file": ("c.json.enc", b"{}")},
|
|
headers=headers,
|
|
)
|
|
assert called["path"] in unlink_calls
|
|
assert not called.get("sync", False)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_vault_import_upload_too_large(client):
|
|
cl, token = client
|
|
called = {}
|
|
api.app.state.pm.handle_import_database = lambda path: called.setdefault(
|
|
"import_called", True
|
|
)
|
|
api.app.state.pm.sync_vault = lambda: called.setdefault("sync_called", True)
|
|
|
|
old_limit = api._MAX_IMPORT_BYTES
|
|
api._MAX_IMPORT_BYTES = 8
|
|
try:
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
res = await cl.post(
|
|
"/api/v1/vault/import",
|
|
files={"file": ("big.json.enc", b"123456789")},
|
|
headers=headers,
|
|
)
|
|
finally:
|
|
api._MAX_IMPORT_BYTES = old_limit
|
|
|
|
assert res.status_code == 413
|
|
assert "exceeds max size" in res.json()["detail"]
|
|
assert called == {}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_vault_import_invalid_extension(client):
|
|
cl, token = client
|
|
api.app.state.pm.handle_import_database = lambda path: None
|
|
api.app.state.pm.sync_vault = lambda: None
|
|
api.app.state.pm.encryption_manager = SimpleNamespace(
|
|
resolve_relative_path=lambda p: p
|
|
)
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
res = await cl.post(
|
|
"/api/v1/vault/import",
|
|
json={"path": "bad.txt"},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 400
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_vault_import_path_traversal_blocked(client, tmp_path):
|
|
cl, token = client
|
|
key = base64.urlsafe_b64encode(os.urandom(32))
|
|
api.app.state.pm.encryption_manager = EncryptionManager(key, tmp_path)
|
|
api.app.state.pm.handle_import_database = lambda path: None
|
|
api.app.state.pm.sync_vault = lambda: None
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
res = await cl.post(
|
|
"/api/v1/vault/import",
|
|
json={"path": "../evil.json.enc"},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 400
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_vault_lock_endpoint(client):
|
|
cl, token = client
|
|
called = {}
|
|
|
|
def lock():
|
|
called["locked"] = True
|
|
api.app.state.pm.locked = True
|
|
|
|
api.app.state.pm.lock_vault = lock
|
|
api.app.state.pm.locked = False
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
res = await cl.post("/api/v1/vault/lock", headers=headers)
|
|
assert res.status_code == 200
|
|
assert res.json() == {"status": "locked"}
|
|
assert called.get("locked") is True
|
|
assert api.app.state.pm.locked is True
|
|
api.app.state.pm.unlock_vault = lambda pw: setattr(
|
|
api.app.state.pm, "locked", False
|
|
)
|
|
api.app.state.pm.unlock_vault("pw")
|
|
assert api.app.state.pm.locked is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_vault_unlock_endpoint(client):
|
|
cl, token = client
|
|
called = {}
|
|
|
|
def unlock(pw):
|
|
called["password"] = pw
|
|
api.app.state.pm.locked = False
|
|
api.app.state.pm.is_locked = False
|
|
return 0.123
|
|
|
|
api.app.state.pm.verify_password = lambda pw: pw == "pw"
|
|
api.app.state.pm.unlock_vault = unlock
|
|
api.app.state.pm.locked = True
|
|
api.app.state.pm.is_locked = True
|
|
|
|
headers = {"Authorization": f"Bearer {token}", "X-SeedPass-Password": "pw"}
|
|
res = await cl.post("/api/v1/vault/unlock", headers=headers)
|
|
assert res.status_code == 200
|
|
assert res.json()["status"] == "unlocked"
|
|
assert called["password"] == "pw"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_entry_endpoints_blocked_when_vault_locked(client):
|
|
cl, token = client
|
|
api.app.state.pm.locked = True
|
|
api.app.state.pm.is_locked = True
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
res = await cl.get("/api/v1/entry", params={"query": "x"}, headers=headers)
|
|
assert res.status_code == 423
|
|
assert res.json()["detail"] == "Vault is locked"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_generate_password_blocked_when_vault_locked(client):
|
|
cl, token = client
|
|
api.app.state.pm.locked = True
|
|
api.app.state.pm.is_locked = True
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
res = await cl.post("/api/v1/password", headers=headers, json={"length": 12})
|
|
assert res.status_code == 423
|
|
assert res.json()["detail"] == "Vault is locked"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_config_endpoint_blocked_when_vault_locked(client):
|
|
cl, token = client
|
|
api.app.state.pm.locked = True
|
|
api.app.state.pm.is_locked = True
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
res = await cl.get("/api/v1/config/k", headers=headers)
|
|
assert res.status_code == 423
|
|
assert res.json()["detail"] == "Vault is locked"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_get_config_denies_sensitive_keys(client):
|
|
cl, token = client
|
|
api.app.state.pm.locked = False
|
|
api.app.state.pm.is_locked = False
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
res = await cl.get("/api/v1/config/password_hash", headers=headers)
|
|
assert res.status_code == 403
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_lock_unlock_cycle_restores_entry_access(client):
|
|
cl, token = client
|
|
api.app.state.pm.locked = False
|
|
api.app.state.pm.is_locked = False
|
|
api.app.state.pm.verify_password = lambda pw: pw == "pw"
|
|
api.app.state.pm.lock_vault = lambda: (
|
|
setattr(api.app.state.pm, "locked", True),
|
|
setattr(api.app.state.pm, "is_locked", True),
|
|
)
|
|
api.app.state.pm.unlock_vault = lambda pw: (
|
|
setattr(api.app.state.pm, "locked", False),
|
|
setattr(api.app.state.pm, "is_locked", False),
|
|
0.01,
|
|
)[-1]
|
|
|
|
auth = {"Authorization": f"Bearer {token}"}
|
|
locked_res = await cl.post("/api/v1/vault/lock", headers=auth)
|
|
assert locked_res.status_code == 200
|
|
|
|
denied = await cl.get("/api/v1/entry", params={"query": "x"}, headers=auth)
|
|
assert denied.status_code == 423
|
|
|
|
unlocked = await cl.post(
|
|
"/api/v1/vault/unlock",
|
|
headers={**auth, "X-SeedPass-Password": "pw"},
|
|
)
|
|
assert unlocked.status_code == 200
|
|
|
|
allowed = await cl.get("/api/v1/entry", params={"query": "x"}, headers=auth)
|
|
assert allowed.status_code == 200
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_secret_mode_endpoint(client):
|
|
cl, token = client
|
|
called = {}
|
|
|
|
def set_secret(val):
|
|
called.setdefault("enabled", val)
|
|
|
|
def set_delay(val):
|
|
called.setdefault("delay", val)
|
|
|
|
api.app.state.pm.config_manager.set_secret_mode_enabled = set_secret
|
|
api.app.state.pm.config_manager.set_clipboard_clear_delay = set_delay
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
res = await cl.post(
|
|
"/api/v1/secret-mode",
|
|
json={"enabled": True, "delay": 12},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
assert res.json() == {"status": "ok"}
|
|
assert called["enabled"] is True
|
|
assert called["delay"] == 12
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_vault_export_endpoint(client, tmp_path):
|
|
cl, token = client
|
|
out = tmp_path / "out.json"
|
|
out.write_text("data")
|
|
|
|
api.app.state.pm.handle_export_database = lambda *a, **k: out
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"X-SeedPass-Password": "pw",
|
|
}
|
|
res = await cl.post("/api/v1/vault/export", headers=headers)
|
|
assert res.status_code == 200
|
|
assert res.content == b"data"
|
|
|
|
res = await cl.post(
|
|
"/api/v1/vault/export", headers={"Authorization": f"Bearer {token}"}
|
|
)
|
|
assert res.status_code == 401
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_backup_parent_seed_endpoint(client, tmp_path):
|
|
cl, token = client
|
|
api.app.state.pm.parent_seed = "seed"
|
|
called = {}
|
|
api.app.state.pm.encryption_manager = SimpleNamespace(
|
|
encrypt_and_save_file=lambda data, path: called.setdefault("path", path),
|
|
resolve_relative_path=lambda p: p,
|
|
)
|
|
path = Path("seed.enc")
|
|
headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"X-SeedPass-Password": "pw",
|
|
}
|
|
res = await cl.post(
|
|
"/api/v1/vault/backup-parent-seed",
|
|
json={"path": str(path), "confirm": True},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
assert res.json() == {"status": "saved", "path": str(path)}
|
|
assert called["path"] == path
|
|
|
|
res = await cl.post(
|
|
"/api/v1/vault/backup-parent-seed",
|
|
json={"path": str(path)},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 400
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_backup_parent_seed_path_traversal_blocked(client, tmp_path):
|
|
cl, token = client
|
|
api.app.state.pm.parent_seed = "seed"
|
|
key = base64.urlsafe_b64encode(os.urandom(32))
|
|
api.app.state.pm.encryption_manager = EncryptionManager(key, tmp_path)
|
|
headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"X-SeedPass-Password": "pw",
|
|
}
|
|
res = await cl.post(
|
|
"/api/v1/vault/backup-parent-seed",
|
|
json={"path": "../evil.enc", "confirm": True},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 400
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_relay_management_endpoints(client, dummy_nostr_client, monkeypatch):
|
|
cl, token = client
|
|
nostr_client, _ = dummy_nostr_client
|
|
relays = ["wss://a", "wss://b"]
|
|
|
|
def load_config(require_pin=False):
|
|
return {"relays": relays.copy()}
|
|
|
|
called = {}
|
|
|
|
def set_relays(new, require_pin=False):
|
|
called["set"] = new
|
|
|
|
api.app.state.pm.config_manager.load_config = load_config
|
|
api.app.state.pm.config_manager.set_relays = set_relays
|
|
monkeypatch.setattr(
|
|
NostrClient,
|
|
"initialize_client_pool",
|
|
lambda self: called.setdefault("init", True),
|
|
)
|
|
monkeypatch.setattr(
|
|
nostr_client, "close_client_pool", lambda: called.setdefault("close", True)
|
|
)
|
|
api.app.state.pm.nostr_client = nostr_client
|
|
api.app.state.pm.nostr_client.relays = relays.copy()
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
res = await cl.get("/api/v1/relays", headers=headers)
|
|
assert res.status_code == 200
|
|
assert res.json() == {"relays": relays}
|
|
|
|
res = await cl.post("/api/v1/relays", json={"url": "wss://c"}, headers=headers)
|
|
assert res.status_code == 200
|
|
assert called["set"] == ["wss://a", "wss://b", "wss://c"]
|
|
|
|
api.app.state.pm.config_manager.load_config = lambda require_pin=False: {
|
|
"relays": ["wss://a", "wss://b", "wss://c"]
|
|
}
|
|
res = await cl.delete("/api/v1/relays/2", headers=headers)
|
|
assert res.status_code == 200
|
|
assert called["set"] == ["wss://a", "wss://c"]
|
|
|
|
res = await cl.post("/api/v1/relays/reset", headers=headers)
|
|
assert res.status_code == 200
|
|
assert called.get("init") is True
|
|
assert api.app.state.pm.nostr_client.relays == list(DEFAULT_RELAYS)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_generate_password_no_special_chars(client):
|
|
cl, token = client
|
|
|
|
class DummyEnc:
|
|
def derive_seed_from_mnemonic(self, mnemonic):
|
|
return b"\x00" * 32
|
|
|
|
class DummyBIP85:
|
|
def derive_entropy(
|
|
self, index: int, entropy_bytes: int, app_no: int = 32
|
|
) -> bytes:
|
|
return bytes(range(entropy_bytes))
|
|
|
|
api.app.state.pm.password_generator = PasswordGenerator(
|
|
DummyEnc(), "seed", DummyBIP85()
|
|
)
|
|
api.app.state.pm.parent_seed = "seed"
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
res = await cl.post(
|
|
"/api/v1/password",
|
|
json={"length": 16, "include_special_chars": False},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
pw = res.json()["password"]
|
|
assert not any(c in string.punctuation for c in pw)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_generate_password_allowed_chars(client):
|
|
cl, token = client
|
|
|
|
class DummyEnc:
|
|
def derive_seed_from_mnemonic(self, mnemonic):
|
|
return b"\x00" * 32
|
|
|
|
class DummyBIP85:
|
|
def derive_entropy(
|
|
self, index: int, entropy_bytes: int, app_no: int = 32
|
|
) -> bytes:
|
|
return bytes((index + i) % 256 for i in range(entropy_bytes))
|
|
|
|
api.app.state.pm.password_generator = PasswordGenerator(
|
|
DummyEnc(), "seed", DummyBIP85()
|
|
)
|
|
api.app.state.pm.parent_seed = "seed"
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
allowed = "@$"
|
|
res = await cl.post(
|
|
"/api/v1/password",
|
|
json={"length": 16, "allowed_special_chars": allowed},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
pw = res.json()["password"]
|
|
specials = [c for c in pw if c in string.punctuation]
|
|
assert specials and all(c in allowed for c in specials)
|