diff --git a/client/src/cmbuyer_client/localstate/models.py b/client/src/cmbuyer_client/localstate/models.py index 5720b6a..8de11d2 100644 --- a/client/src/cmbuyer_client/localstate/models.py +++ b/client/src/cmbuyer_client/localstate/models.py @@ -51,6 +51,18 @@ class LoadedProfile: return f"LoadedProfile(settings={self.settings!r}, credentials=[已隐藏])" +@dataclass(frozen=True) +class ProfileSummary: + """不解密、不返回任何 token 数据的配置页只读摘要。""" + + settings: ProfileSettings + has_stored_device_token: bool + + def __post_init__(self) -> None: + if type(self.has_stored_device_token) is not bool: + raise ValueError("invalid_token_presence") + + @dataclass(frozen=True) class PollingSession: profile_id: str diff --git a/client/src/cmbuyer_client/localstate/store.py b/client/src/cmbuyer_client/localstate/store.py index fe43a75..a91bc68 100644 --- a/client/src/cmbuyer_client/localstate/store.py +++ b/client/src/cmbuyer_client/localstate/store.py @@ -31,7 +31,7 @@ from cmbuyer_client.core.models import ( ) from cmbuyer_client.core.validation import datetime_nanoseconds, require_rfc3339_z, require_uuid4, rfc3339_z_nanoseconds -from .models import LoadedProfile, PendingEvidence, PollingSession, ProfileSettings, RecoverySnapshot +from .models import LoadedProfile, PendingEvidence, PollingSession, ProfileSettings, ProfileSummary, RecoverySnapshot from .protection import SecretProtector @@ -346,6 +346,22 @@ class LocalStateStore: token = self._unprotect_token(bytes(row[2]), purpose=_device_token_purpose(profile_id, settings.device_id)) return LoadedProfile(settings, DeviceCredentials(settings.device_id, token)) + def load_profile_summary(self, profile_id: str) -> ProfileSummary: + """读取非秘密设置与 token 存在性,绝不经过 SecretProtector。""" + + with self._read_transaction() as connection: + row = connection.execute( + """SELECT service_url,device_id,adb_path,adb_serial,transport, + poll_interval_seconds,failure_threshold,http_timeout_seconds,step_timeout_seconds, + CASE WHEN length(device_token_cipher) > 0 THEN 1 ELSE 0 END + FROM profiles WHERE profile_id=?""", + (profile_id,), + ).fetchone() + if row is None: + raise StateError("profile_not_found") + settings = ProfileSettings(profile_id, *row[:9]) + return ProfileSummary(settings, bool(row[9])) + def start_or_resume_polling(self, profile_id: str) -> PollingSession: now = self._utc_now() with self._transaction() as connection: diff --git a/client/tests/localstate/test_store.py b/client/tests/localstate/test_store.py index bbc3668..7b8c595 100644 --- a/client/tests/localstate/test_store.py +++ b/client/tests/localstate/test_store.py @@ -112,6 +112,30 @@ class LocalStateStoreTests(unittest.TestCase): finally: os.chdir(original_cwd) + def test_profile_summary_reads_settings_and_token_presence_without_unprotect(self) -> None: + class NoUnprotectProtector(FakeProtector): + def unprotect(self, ciphertext: bytes, *, purpose: str) -> bytes: + raise AssertionError("metadata_read_must_not_unprotect") + + summary_store = LocalStateStore( + self.database, + NoUnprotectProtector(), + now=lambda: self.clock[0], + ) + summary = summary_store.load_profile_summary(PROFILE) + self.assertEqual(summary.settings, settings()) + self.assertTrue(summary.has_stored_device_token) + self.assertNotIn(DEVICE_TOKEN, repr(summary)) + + def test_profile_summary_missing_profile_fails_without_creating_defaults(self) -> None: + with self.assertRaisesRegex(StateError, "profile_not_found"): + self.store.load_profile_summary("missing") + connection = sqlite3.connect(self.database) + try: + self.assertEqual(connection.execute("SELECT count(*) FROM profiles").fetchone()[0], 1) + finally: + connection.close() + def test_empty_allows_new_key_but_terminal_does_not(self) -> None: self.store.start_or_resume_polling(PROFILE) first = self.store.prepare_claim(PROFILE)