feat(client): expose secret-free profile summary
This commit is contained in:
@@ -51,6 +51,18 @@ class LoadedProfile:
|
|||||||
return f"LoadedProfile(settings={self.settings!r}, credentials=[已隐藏])"
|
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)
|
@dataclass(frozen=True)
|
||||||
class PollingSession:
|
class PollingSession:
|
||||||
profile_id: str
|
profile_id: str
|
||||||
|
|||||||
@@ -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 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
|
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))
|
token = self._unprotect_token(bytes(row[2]), purpose=_device_token_purpose(profile_id, settings.device_id))
|
||||||
return LoadedProfile(settings, DeviceCredentials(settings.device_id, token))
|
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:
|
def start_or_resume_polling(self, profile_id: str) -> PollingSession:
|
||||||
now = self._utc_now()
|
now = self._utc_now()
|
||||||
with self._transaction() as connection:
|
with self._transaction() as connection:
|
||||||
|
|||||||
@@ -112,6 +112,30 @@ class LocalStateStoreTests(unittest.TestCase):
|
|||||||
finally:
|
finally:
|
||||||
os.chdir(original_cwd)
|
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:
|
def test_empty_allows_new_key_but_terminal_does_not(self) -> None:
|
||||||
self.store.start_or_resume_polling(PROFILE)
|
self.store.start_or_resume_polling(PROFILE)
|
||||||
first = self.store.prepare_claim(PROFILE)
|
first = self.store.prepare_claim(PROFILE)
|
||||||
|
|||||||
Reference in New Issue
Block a user