"""Verify the actual phone/watch chain and save a report without bearer tokens or raw watch logs.""" import argparse import hashlib import json from pathlib import Path import sys import time import urllib.error import urllib.request from datetime import datetime, timezone, timedelta from bridge import Bridge def safe_watch_info(result): """Keep explicit model/version fields even when the service nests values as JSON.""" allowed = {'brand', 'model', 'deviceModel', 'device_model', 'version', 'runtime', 'runtimeVersion', 'runtime_version', 'nucleus', 'nucleusVersion', 'nucleus_version', 'firmwareVersion', 'firmware_version', 'osVersion', 'osVersionName', 'masVersion'} safe = {} pending = [result] for _ in range(12): if not pending: break result = pending.pop(0) if isinstance(result, str): try: result = json.loads(result) except (ValueError, TypeError): continue if not isinstance(result, dict): continue for key, value in result.items(): if key in allowed and isinstance(value, (str, int, float)): safe[key] = value # OWW265 puts its JSON object in msg while values is an empty string. # Only traverse these two envelopes and retain the same explicit fields. pending.extend(result.get(key) for key in ('values', 'msg') if key in result) return safe def watch_success(job): result = job.get('result') return job.get('state') == 'success' and (not isinstance(result, dict) or result.get('code', 200) == 200) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--serial', required=True) parser.add_argument('--probe', type=Path, help='Built com.sydxky.rtos.deviceprobe bundle.pkg; installs this package on the watch') parser.add_argument('--output', type=Path, required=True) parser.add_argument('--config', type=Path, help='Local appId/license JSON; never copied to report') parser.add_argument('--ready-wait', type=int, default=30, help='Seconds to wait for the probe ready log (1-300)') args = parser.parse_args() if not 1 <= args.ready_wait <= 300: parser.error('--ready-wait must be 1-300 seconds') report = {'time': datetime.now(timezone(timedelta(hours=8))).isoformat(), 'steps': []} bridge = Bridge(args.serial) def record(name, passed, evidence): report['steps'].append({'name': name, 'passed': bool(passed), 'evidence': evidence}) print(f'{name}: {"PASS" if passed else "FAIL"}', flush=True) try: bridge.prepare() phone = bridge.shell('shell', 'getprop', 'ro.product.model') api = bridge.shell('shell', 'getprop', 'ro.build.version.sdk') report['phone'] = {'model': phone, 'androidApi': api} record('phone_api', True, bridge.request('GET', '/status')) for name, headers in [('missing_token', {}), ('browser_origin', {'Authorization': 'Bearer ' + bridge.token, 'Origin': 'http://example.invalid'})]: request = urllib.request.Request('http://127.0.0.1:18765/status', headers=headers) try: with bridge.opener.open(request, timeout=5) as response: code = response.status except urllib.error.HTTPError as error: code = error.code record(name, code == 401, {'http': code}) config = json.loads(args.config.read_text(encoding='utf-8')) if args.config else {} try: status = bridge.connect(config.get('appId', 'test'), config.get('license', 'test'), config.get('service', 'com.heytap.health')) except Exception: status = bridge.request('GET', '/status') record('wear_engine_binding', False, {'state': status.get('state')}) raise record('wear_engine_binding', status['state'] == 'bound', status) link = bridge.wait_link(45) connected = watch_success(link) and link.get('result', {}).get('connected') is True record('watch_connected', connected, link) if not connected: raise RuntimeError('Watch is offline; wake it before testing installation') info = bridge.command({'command': 'info'}, 30) # Only retain software/model fields; do not publish device identifiers from the callback. record('watch_info_callback', watch_success(info), {'state': info['state'], 'result': safe_watch_info(info.get('result'))}) if args.probe: report['probe_sha256'] = hashlib.sha256(args.probe.read_bytes()).hexdigest() cursor = max((row['seq'] for row in bridge.request('GET', '/events?after=0')['events']), default=0) logs = bridge.command({'command': 'logs', 'enabled': True}, 30) record('rpc_logs_enable', watch_success(logs), logs) try: upload = bridge.upload(args.probe) install = bridge.command({'command': 'install', 'file': upload}) record('watch_install_callback', watch_success(install), install) if watch_success(install): launch = bridge.command({'command': 'launch', 'package': 'com.sydxky.rtos.deviceprobe'}, 30) record('watch_launch_callback', watch_success(launch), launch) markers = [] deadline = time.monotonic() + args.ready_wait while time.monotonic() < deadline: rows = bridge.request('GET', '/events?after=' + str(cursor))['events'] for row in rows: cursor = row['seq'] if row['type'] == 'watch.log' and 'RTOS_DEVICE_PROBE_' in str(row['data']): # The marker carries no user content. Omit the rest of the log envelope. text = str(row['data']) for marker in ('RTOS_DEVICE_PROBE_APP_CREATE_V1', 'RTOS_DEVICE_PROBE_APP_SHOW_V1', 'RTOS_DEVICE_PROBE_READY_V1'): if marker in text and marker not in markers: markers.append(marker) if 'RTOS_DEVICE_PROBE_READY_V1' in markers: break time.sleep(.5) record('watch_page_ready_log', 'RTOS_DEVICE_PROBE_READY_V1' in markers, markers) finally: off = bridge.command({'command': 'logs', 'enabled': False}, 20) record('rpc_logs_disable', watch_success(off), off) report['passed'] = all(step['passed'] for step in report['steps']) except Exception as error: report['passed'] = False # Never print config, token or raw transport payloads here. report['error'] = type(error).__name__ print('Verification incomplete: ' + type(error).__name__, file=sys.stderr) finally: args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2) + '\n', encoding='utf-8') return 0 if report['passed'] else 2 if __name__ == '__main__': sys.stdout.reconfigure(encoding='utf-8') sys.stderr.reconfigure(encoding='utf-8') sys.exit(main())