From e7f41d6e27b4cdba12cbdcb869f9a396f9195fff Mon Sep 17 00:00:00 2001 From: chendelian <116870791@qq.com> Date: Mon, 11 May 2026 10:51:29 +0800 Subject: [PATCH] test(crypto+db+cli): add master key, crypto, credentials_repo, and CLI tests Co-authored-by: Cursor --- tests/conftest.py | 7 ++ tests/test_cli_master_key.py | 150 +++++++++++++++++++++++++++++ tests/test_credentials_repo.py | 143 +++++++++++++++++++++++++++ tests/test_crypto.py | 47 +++++++++ tests/test_master_key.py | 60 ++++++++++++ tests/test_schema_migration.py | 3 +- tests/test_schema_migration_v21.py | 104 ++++++++++++++++++++ 7 files changed, 513 insertions(+), 1 deletion(-) create mode 100644 tests/test_cli_master_key.py create mode 100644 tests/test_credentials_repo.py create mode 100644 tests/test_crypto.py create mode 100644 tests/test_master_key.py create mode 100644 tests/test_schema_migration_v21.py diff --git a/tests/conftest.py b/tests/conftest.py index bb2cd2f..9e76834 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -15,6 +15,13 @@ _scripts = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__ if _scripts not in sys.path: sys.path.insert(0, _scripts) +try: + import cryptography # noqa: F401 +except ImportError as e: + raise RuntimeError( + "Stage B requires 'cryptography'. pip install cryptography --break-system-packages" + ) from e + import pytest diff --git a/tests/test_cli_master_key.py b/tests/test_cli_master_key.py new file mode 100644 index 0000000..f2df7f3 --- /dev/null +++ b/tests/test_cli_master_key.py @@ -0,0 +1,150 @@ +# -*- coding: utf-8 -*- +import json +import os +import subprocess +import sys +import tempfile +from dataclasses import dataclass + +import pytest + +_scripts = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts") +if _scripts not in sys.path: + sys.path.insert(0, _scripts) + +MAIN_PY = os.path.join(_scripts, "main.py") + + +def _iso_cli_env(data_root: str, uid: str = "cli_u") -> dict: + """Align skill roots so subprocess + in-process tests hit the same DB.""" + return { + "JIANGCHANG_DATA_ROOT": data_root, + "JIANGCHANG_USER_ID": uid, + "CLAW_DATA_ROOT": data_root, + "CLAW_USER_ID": uid, + } + + +@dataclass +class _ProcOut: + returncode: int + stdout: str + stderr: str + + +def _run_cli(env: dict, args: list[str], stdin: str | None = None) -> _ProcOut: + merged = {**os.environ, **env} + merged.pop("ACCOUNT_MANAGER_MASTER_KEY", None) + inp = stdin.encode("utf-8") if stdin is not None else None + r = subprocess.run( + [sys.executable, MAIN_PY, *args], + input=inp, + capture_output=True, + env=merged, + cwd=os.path.dirname(_scripts), + ) + stdout_txt = r.stdout.decode("utf-8") if r.stdout else "" + stderr_txt = r.stderr.decode("utf-8", errors="replace") if r.stderr else "" + return _ProcOut(returncode=r.returncode, stdout=stdout_txt, stderr=stderr_txt) + + +def test_init_master_key_writes_file(monkeypatch): + tmp = tempfile.mkdtemp(prefix="cli_mk_") + data_root = os.path.join(tmp, "dr") + os.makedirs(data_root) + env = _iso_cli_env(data_root) + proc = _run_cli(env, ["account", "init-master-key"]) + assert proc.returncode == 0 + payload = json.loads(proc.stdout.strip().splitlines()[0]) + assert payload["success"] is True + assert os.path.isfile(payload["path"]) + + +def test_init_master_key_rejects_duplicate(monkeypatch): + tmp = tempfile.mkdtemp(prefix="cli_mk_d_") + data_root = os.path.join(tmp, "dr") + os.makedirs(data_root) + env = _iso_cli_env(data_root) + p1 = _run_cli(env, ["account", "init-master-key"]) + assert p1.returncode == 0 + p2 = _run_cli(env, ["account", "init-master-key"]) + assert p2.returncode == 0 + payload = json.loads(p2.stdout.strip().splitlines()[0]) + assert payload["success"] is False + assert payload["error"]["code"] == "ERR_MASTER_KEY_EXISTS" + + +def test_export_import_roundtrip(monkeypatch): + tmp = tempfile.mkdtemp(prefix="cli_ex_") + data_root = os.path.join(tmp, "dr") + os.makedirs(data_root) + env = _iso_cli_env(data_root) + + init = _run_cli(env, ["account", "init-master-key"]) + assert init.returncode == 0 + + # Insert encrypted credential via repo in-process (same paths as CLI) + monkeypatch.setenv("JIANGCHANG_DATA_ROOT", data_root) + monkeypatch.setenv("JIANGCHANG_USER_ID", "cli_u") + monkeypatch.setenv("CLAW_DATA_ROOT", data_root) + monkeypatch.setenv("CLAW_USER_ID", "cli_u") + monkeypatch.delenv("ACCOUNT_MANAGER_MASTER_KEY", raising=False) + + from util.constants import SKILL_SLUG + from util.logging_config import setup_skill_logging + + setup_skill_logging(SKILL_SLUG, "openclaw.skill.account_manager_cli_inline") + + from db.connection import get_conn, init_db + from db.credentials_repo import insert_credential_encrypted, read_credential_plaintext + + init_db() + conn = get_conn() + try: + cid = insert_credential_encrypted( + conn, + account_id=None, + platform_key="icbc_sim", + credential_label="ICBC password label", + credential_type="password", + plaintext="roundtrip-secret", + expires_at=None, + extra_json="{}", + now=1_750_000_000, + ) + conn.commit() + finally: + conn.close() + + conn = get_conn() + try: + cur = conn.cursor() + cur.execute("SELECT COUNT(*) FROM credentials") + assert int(cur.fetchone()[0]) == 1 + finally: + conn.close() + + ex = _run_cli(env, ["account", "export-credentials", "--plaintext"]) + assert ex.returncode == 0 + arr = json.loads(ex.stdout.strip()) + assert len(arr) == 1 + assert arr[0]["plaintext"] == "roundtrip-secret" + + imp = _run_cli( + env, + ["account", "import-credentials", "--replace-all"], + stdin=json.dumps(arr, ensure_ascii=False), + ) + assert imp.returncode == 0 + + conn = get_conn() + try: + cur = conn.cursor() + cur.execute("SELECT COUNT(*) FROM credentials WHERE secret_storage='local_encrypted'") + assert int(cur.fetchone()[0]) == 1 + cur.execute("SELECT id FROM credentials LIMIT 1") + new_id = int(cur.fetchone()[0]) + pt = read_credential_plaintext(conn, new_id) + assert pt == "roundtrip-secret" + finally: + conn.close() diff --git a/tests/test_credentials_repo.py b/tests/test_credentials_repo.py new file mode 100644 index 0000000..ba9b83e --- /dev/null +++ b/tests/test_credentials_repo.py @@ -0,0 +1,143 @@ +# -*- coding: utf-8 -*- +import pytest +from cryptography.fernet import Fernet + +from db.accounts_repo import insert_credential + + +class TestCredentialsRepo: + def test_encrypted_insert_and_read(self, monkeypatch, clean_db): + from db.credentials_repo import ( + insert_credential_encrypted, + list_credentials_by_account, + read_credential_plaintext, + ) + from service.master_key import ENV_MASTER_KEY + + key = Fernet.generate_key() + monkeypatch.setenv(ENV_MASTER_KEY, key.decode("ascii")) + + conn = clean_db + now = 1_700_000_000 + aid = insert_credential_encrypted( + conn, + account_id=7, + platform_key="icbc_sim", + credential_label="工行密码", + credential_type="password", + plaintext="my-password", + expires_at=None, + extra_json="{}", + now=now, + ) + conn.commit() + pt = read_credential_plaintext(conn, aid) + assert pt == "my-password" + assert "my-password" not in "".join( + str(v) for v in list_credentials_by_account(conn, 7) + ) + + def test_insert_validation(self, monkeypatch, clean_db): + from db.credentials_repo import insert_credential_encrypted, insert_credential_external + from service.master_key import ENV_MASTER_KEY + + monkeypatch.setenv(ENV_MASTER_KEY, Fernet.generate_key().decode("ascii")) + conn = clean_db + now = 1 + with pytest.raises(ValueError, match="credential_type"): + insert_credential_encrypted( + conn, + account_id=None, + platform_key="x", + credential_label="l", + credential_type="weird", + plaintext="x", + expires_at=None, + extra_json="{}", + now=now, + ) + with pytest.raises(ValueError, match="plaintext"): + insert_credential_encrypted( + conn, + account_id=None, + platform_key="x", + credential_label="l", + credential_type="password", + plaintext="", + expires_at=None, + extra_json="{}", + now=now, + ) + with pytest.raises(ValueError, match="secret_storage"): + insert_credential_external( + conn, + account_id=None, + platform_key="x", + credential_label="l", + credential_type="password", + secret_storage="bogus", + secret_ref=None, + secret_mask="", + expires_at=None, + extra_json="{}", + now=now, + ) + + def test_list_has_no_sensitive_keys(self, monkeypatch, clean_db): + from db.credentials_repo import insert_credential_encrypted, list_credentials_by_account + from service.master_key import ENV_MASTER_KEY + + monkeypatch.setenv(ENV_MASTER_KEY, Fernet.generate_key().decode("ascii")) + conn = clean_db + now = 2 + insert_credential_encrypted( + conn, + account_id=9, + platform_key="p", + credential_label="a", + credential_type="api_key", + plaintext="k1", + expires_at=None, + extra_json="{}", + now=now, + ) + insert_credential_encrypted( + conn, + account_id=9, + platform_key="p", + credential_label="b", + credential_type="api_token", + plaintext="k2", + expires_at=None, + extra_json="{}", + now=now + 1, + ) + conn.commit() + rows = list_credentials_by_account(conn, 9) + assert len(rows) == 2 + for d in rows: + assert "plaintext" not in d + assert "secret_ciphertext" not in d + + def test_none_storage_read_fails(self, clean_db): + from db.credentials_repo import CredentialNotEncryptedError, read_credential_plaintext + + conn = clean_db + now = 3 + cid = insert_credential( + conn, + account_id=None, + platform_key="p", + credential_label="n", + credential_type="note", + secret_storage="none", + secret_ref=None, + secret_mask="", + expires_at=None, + extra_json="{}", + now=now, + ) + conn.commit() + with pytest.raises(CredentialNotEncryptedError) as ei: + read_credential_plaintext(conn, cid) + assert ei.value.code == "ERR_CREDENTIAL_NOT_ENCRYPTED" diff --git a/tests/test_crypto.py b/tests/test_crypto.py new file mode 100644 index 0000000..d9a70e9 --- /dev/null +++ b/tests/test_crypto.py @@ -0,0 +1,47 @@ +# -*- coding: utf-8 -*- +import os +import sys +import tempfile + +import pytest +from cryptography.fernet import Fernet + +_scripts = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts") +if _scripts not in sys.path: + sys.path.insert(0, _scripts) + +from service.crypto import CredentialDecryptError, decrypt_secret, encrypt_secret # noqa: E402 +from service.master_key import ENV_MASTER_KEY # noqa: E402 + + +@pytest.fixture() +def crypto_env(monkeypatch): + tmp = tempfile.mkdtemp(prefix="crypto_test_") + data_root = os.path.join(tmp, "claw-data") + os.makedirs(data_root, exist_ok=True) + monkeypatch.setenv("JIANGCHANG_DATA_ROOT", data_root) + monkeypatch.setenv("JIANGCHANG_USER_ID", "c_uid") + key = Fernet.generate_key() + monkeypatch.setenv(ENV_MASTER_KEY, key.decode("ascii")) + yield key + + +def test_encrypt_decrypt_roundtrip(crypto_env): + pt = "my-password-123" + ct = encrypt_secret(pt) + assert decrypt_secret(ct) == pt + + +def test_decrypt_invalid_token(crypto_env): + with pytest.raises(CredentialDecryptError) as ei: + decrypt_secret("AAAA-invalid-token-AAAA") + assert ei.value.code == "ERR_CREDENTIAL_DECRYPT_FAILED" + + +def test_decrypt_wrong_master_key(monkeypatch, crypto_env): + pt = "secret-value" + ct = encrypt_secret(pt) + other = Fernet.generate_key().decode("ascii") + monkeypatch.setenv(ENV_MASTER_KEY, other) + with pytest.raises(CredentialDecryptError): + decrypt_secret(ct) diff --git a/tests/test_master_key.py b/tests/test_master_key.py new file mode 100644 index 0000000..92311d1 --- /dev/null +++ b/tests/test_master_key.py @@ -0,0 +1,60 @@ +# -*- coding: utf-8 -*- +import os +import sys +import tempfile + +import pytest +from cryptography.fernet import Fernet + +_scripts = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts") +if _scripts not in sys.path: + sys.path.insert(0, _scripts) + +from service.master_key import ( # noqa: E402 + ENV_MASTER_KEY, + MasterKeyMissingError, + load_master_key, + write_master_key_file, +) + + +@pytest.fixture() +def iso_mk(monkeypatch): + tmp = tempfile.mkdtemp(prefix="mk_test_") + data_root = os.path.join(tmp, "claw-data") + os.makedirs(data_root, exist_ok=True) + monkeypatch.setenv("JIANGCHANG_DATA_ROOT", data_root) + monkeypatch.setenv("JIANGCHANG_USER_ID", "mk_uid") + monkeypatch.delenv(ENV_MASTER_KEY, raising=False) + yield tmp + + +def test_env_priority_over_file(iso_mk, monkeypatch): + key_env = Fernet.generate_key() + key_file = Fernet.generate_key() + monkeypatch.setenv(ENV_MASTER_KEY, key_env.decode("ascii")) + write_master_key_file(key_file, allow_overwrite=True) + assert load_master_key() == key_env + + +def test_file_fallback(iso_mk): + key_file = Fernet.generate_key() + write_master_key_file(key_file, allow_overwrite=True) + assert load_master_key() == key_file + + +def test_missing_raises(iso_mk): + with pytest.raises(MasterKeyMissingError) as ei: + load_master_key() + assert ei.value.code == "ERR_MASTER_KEY_MISSING" + + +def test_invalid_env_falls_back_to_file(iso_mk, monkeypatch): + monkeypatch.setenv(ENV_MASTER_KEY, "not-a-real-key") + with pytest.raises(MasterKeyMissingError): + load_master_key() + + key_file = Fernet.generate_key() + write_master_key_file(key_file, allow_overwrite=True) + monkeypatch.setenv(ENV_MASTER_KEY, "not-a-real-key") + assert load_master_key() == key_file diff --git a/tests/test_schema_migration.py b/tests/test_schema_migration.py index 26a55be..97832ab 100644 --- a/tests/test_schema_migration.py +++ b/tests/test_schema_migration.py @@ -130,13 +130,14 @@ class TestFreshSchema: assert "default_auth_strategy" in cols_plat cols_cred = _column_names(conn, "credentials") assert "account_id" in cols_cred + assert "secret_ciphertext" in cols_cred cur = conn.cursor() cur.execute( "SELECT value FROM _schema_meta WHERE key = ?", ("schema_version",), ) row = cur.fetchone() - assert row is not None and row[0] == "2" + assert row is not None and row[0] == "2.1" finally: conn.close() diff --git a/tests/test_schema_migration_v21.py b/tests/test_schema_migration_v21.py new file mode 100644 index 0000000..d9b8115 --- /dev/null +++ b/tests/test_schema_migration_v21.py @@ -0,0 +1,104 @@ +# -*- coding: utf-8 -*- +import os +import sqlite3 +import sys +import tempfile + +import pytest + +_scripts = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts") +if _scripts not in sys.path: + sys.path.insert(0, _scripts) + +from util.constants import SKILL_SLUG # noqa: E402 + + +def _db_path(root: str, uid: str) -> str: + return os.path.join(root, uid, SKILL_SLUG, "account-manager.db") + + +def _prep_logging(): + from util.logging_config import setup_skill_logging + + setup_skill_logging(SKILL_SLUG, "openclaw.skill.account_manager_schema_v21") + + +def _column_names(conn: sqlite3.Connection, table: str) -> list[str]: + cur = conn.cursor() + cur.execute(f"PRAGMA table_info({table})") + return [row[1] for row in cur.fetchall()] + + +@pytest.fixture() +def iso_v21(monkeypatch): + with tempfile.TemporaryDirectory(prefix="acc_mgr_v21_", ignore_cleanup_errors=True) as tmp: + data_root = os.path.join(tmp, "data-root") + os.makedirs(data_root) + uid = "v21_uid" + monkeypatch.setenv("JIANGCHANG_DATA_ROOT", data_root) + monkeypatch.setenv("JIANGCHANG_USER_ID", uid) + db_path = _db_path(data_root, uid) + os.makedirs(os.path.dirname(db_path), exist_ok=True) + _prep_logging() + yield {"tmp": tmp, "db_path": db_path} + + +class TestSchemaV21: + def test_fresh_has_secret_ciphertext_and_meta(self, iso_v21): + from db.connection import get_conn, init_db + + if os.path.isfile(iso_v21["db_path"]): + os.remove(iso_v21["db_path"]) + init_db() + conn = get_conn() + try: + assert "secret_ciphertext" in _column_names(conn, "credentials") + cur = conn.cursor() + cur.execute( + "SELECT value FROM _schema_meta WHERE key = ?", + ("schema_version",), + ) + assert cur.fetchone()[0] == "2.1" + finally: + conn.close() + + def test_v2_to_v21_idempotent(self, iso_v21, capsys): + from db.connection import get_conn, init_db + + db_path = iso_v21["db_path"] + if os.path.isfile(db_path): + os.remove(db_path) + + init_db() + conn = get_conn() + try: + conn.execute("ALTER TABLE credentials DROP COLUMN secret_ciphertext") + conn.execute( + "UPDATE _schema_meta SET value = ? WHERE key = ?", + ("2", "schema_version"), + ) + conn.commit() + finally: + conn.close() + + capsys.readouterr() + init_db() + err = capsys.readouterr().err + assert "v2→v2.1" in err + + conn = get_conn() + try: + assert "secret_ciphertext" in _column_names(conn, "credentials") + cur = conn.cursor() + cur.execute( + "SELECT value FROM _schema_meta WHERE key = ?", + ("schema_version",), + ) + assert cur.fetchone()[0] == "2.1" + finally: + conn.close() + + capsys.readouterr() + init_db() + err2 = capsys.readouterr().err + assert "v2→v2.1" not in err2