63 lines
2.8 KiB
Python
63 lines
2.8 KiB
Python
"""Require a confirmed manual send journal entry before enabling an unverified build.
|
|
|
|
UI flags and exported JSON reports are never accepted as activation evidence.
|
|
"""
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
from runtime_paths import application_data_dir
|
|
|
|
TEST_TEXT = '协议通道测试，请忽略'
|
|
|
|
|
|
def local_send_receipt(profile, account, *, journal_path=None):
|
|
account = str(account or '')
|
|
if not profile or not account.isascii() or not account.isdecimal() or int(account) == 0:
|
|
return None
|
|
path = Path(journal_path) if journal_path else application_data_dir() / 'protocol_sends.sqlite3'
|
|
if not path.is_file():
|
|
return None
|
|
try:
|
|
connection = sqlite3.connect(path.resolve().as_uri() + '?mode=ro', uri=True, timeout=1)
|
|
try:
|
|
rows = connection.execute(
|
|
"SELECT request_id,fingerprint,result FROM sends WHERE status='confirmed' "
|
|
"AND request_id LIKE ? ORDER BY created DESC LIMIT 20",
|
|
(f'manual-verification:%:{profile.version}:{account}:%',)).fetchall()
|
|
finally:
|
|
connection.close()
|
|
for request, fingerprint, encoded in rows:
|
|
if len(encoded) > 16000:
|
|
continue
|
|
try:
|
|
receipt = json.loads(encoded)
|
|
if not isinstance(receipt, dict):
|
|
continue
|
|
conversation = receipt.get('conversationId', '')
|
|
if not isinstance(conversation, str) or not conversation.startswith('S:'):
|
|
continue
|
|
if not request.endswith(f':{profile.version}:{account}:{conversation}'):
|
|
continue
|
|
if any(receipt.get(key) != value for key, value in (
|
|
('status', 'confirmed'), ('nativeCallback', 'success'), ('accountId', account),
|
|
('clientVersion', profile.version), ('clientSha256', profile.sha256),
|
|
('adapterRevision', profile.adapter_revision))):
|
|
continue
|
|
if receipt.get('senderVerified') is not True or receipt.get('textVerified') is not True:
|
|
continue
|
|
if any(type(receipt.get(key)) is not int or receipt[key] <= 0 for key in ('messageId', 'serverId')):
|
|
continue
|
|
expected = hashlib.sha256(json.dumps([account, conversation, TEST_TEXT], ensure_ascii=False).encode()).hexdigest()
|
|
if fingerprint == expected:
|
|
return receipt
|
|
except (ValueError, TypeError):
|
|
continue
|
|
except (OSError, sqlite3.Error, ValueError):
|
|
return None
|
|
return None
|
|
|
|
|
|
def send_authorized(profile, account):
|
|
return bool(profile and (profile.verified or local_send_receipt(profile, account)))
|