DroidDeck/tools/pixel_reset_diagnostics.py

#!/usr/bin/env python3
"""Read-only reboot projection. Never writes raw Android logs or account data.

This is evidence collection, not a GPU fault injector or a root-cause oracle.
The optional kernel projection reads only SYSTEM_LAST_KMSG after user unlock.
It emits fixed categories and numeric hardware fields, never arbitrary log lines.
"""
import argparse
from datetime import datetime, timezone
import json
import os
from pathlib import Path
import re
import subprocess
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

ENTRY = re.compile(r'^(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\s+(SYSTEM_LAST_KMSG)\b', re.M)
CATEGORIES = ('GNOC', 'PVRSRV', 'watchdog', 'IOMMU', 'S2MPU', 'access control', 'Kernel panic', 'HWR', 'GPU fault')
PRIMARY_USER_STATES = frozenset(('RUNNING_LOCKED', 'RUNNING_UNLOCKING', 'RUNNING_UNLOCKED'))
# Exact canonical reasons plus the vendor reason observed in the bounded Pixel
# tests. Unrecognized/custom suffixes are not diagnostic output categories.
BOOT_REASONS = frozenset(('cold', 'hard', 'warm', 'reboot', 'shutdown',
                          'recovery', 'bootloader', 'kernel_panic', 'watchdog',
                          'reset,warm,gnoc_error'))


def primary_state(text):
    match = re.search(r'Started users state: \[0=([A-Z_]+)(?=[,\]])', text)
    return match.group(1) if match and match.group(1) in PRIMARY_USER_STATES else 'unknown'


def safe_reason(text):
    value = text.strip()
    return value if value in BOOT_REASONS else 'unknown'


def kernel_projection(text, device_timezone, after=None):
    entries = list(ENTRY.finditer(text))
    result = {'metadata_dates': [m.group(1) for m in entries], 'raw_body_saved': False,
              'root_cause_proven': False}
    if not entries:
        result['status'] = 'no_accessible_entry'
        return result
    latest = max(enumerate(entries), key=lambda row: row[1].group(1))
    index, match = latest
    body = text[match.end():entries[index + 1].start() if index + 1 < len(entries) else len(text)]
    result['entry_date_device_local'] = match.group(1)
    try:
        naive = datetime.strptime(match.group(1), '%Y-%m-%d %H:%M:%S')
        zone = ZoneInfo(device_timezone)
    except (ValueError, ZoneInfoNotFoundError):
        result['status'] = 'unknown_device_timezone'
        return result
    # Dropbox supplies a local wall clock without an offset. Round-trip both
    # folds: a repeated hour has two possible instants; a skipped hour has none.
    instants = {local.astimezone(timezone.utc)
                for fold in (0, 1)
                for local in (naive.replace(tzinfo=zone, fold=fold),)
                if local.astimezone(timezone.utc).astimezone(zone).replace(tzinfo=None) == naive}
    if len(instants) != 1:
        result['status'] = ('ambiguous_device_local_time' if instants
                            else 'nonexistent_device_local_time')
        return result
    local = instants.pop()
    result['entry_date_utc'] = local.isoformat()
    if after is not None and local < after:
        result['status'] = 'stale_entry_not_attributed_to_current_trial'
        return result
    result['status'] = 'projected_entry_not_a_root_cause'
    result['categories'] = {key: len(re.findall(re.escape(key), body, re.I)) for key in CATEGORIES}
    patterns = {
        'gnoc_addresses': r'Error from GNOC at addr (0x[0-9a-fA-F]+)',
        'gnoc_error_types': r'GNOC error type (0x[0-9a-fA-F]+)',
        'firmware_reboot_codes': r'Reboot reason:\s+(0x[0-9a-fA-F]+)',
        'access_control_addresses': r'Access Control Error \((0x[0-9a-fA-F]+)',
    }
    result['hardware_fields'] = {key: sorted(set(re.findall(pattern, body))) for key, pattern in patterns.items()}
    # Read the complete value, not just a recognized-looking prefix. Apply the
    # same fixed projection as getprop; no arbitrary kernel reason text escapes.
    reasons = re.findall(r'Last boot reason:[ \t]*([^\r\n]*)', body)
    result['last_boot_reasons'] = sorted({safe_reason(reason) for reason in reasons})
    return result


def collect(adb, serial, include_kernel=False, after=None):
    def shell(*args):
        try:
            process = subprocess.run([adb, '-s', serial, 'shell', *args], capture_output=True,
                                     text=True, timeout=20)
        except (OSError, subprocess.TimeoutExpired):
            return None, ''
        # No command stderr is copied to output: it may contain unrelated data.
        return process.returncode, process.stdout if process.returncode == 0 else ''

    result = {'at': datetime.now(timezone.utc).isoformat(), 'read_only': True,
              'raw_logs_saved': False, 'credentials_collected': False}
    code, value = shell('settings', 'get', 'global', 'boot_count')
    result['boot_count'] = int(value.strip()) if code == 0 and value.strip().isdigit() else None
    code, value = shell('getprop', 'ro.boot.bootreason')
    result['bootloader_reason'] = safe_reason(value)
    code, value = shell('getprop', 'sys.boot.reason')
    result['system_boot_reason'] = safe_reason(value)
    code, value = shell('dumpsys', 'user')
    result['primary_user_state'] = primary_state(value)
    code, value = shell('getprop', 'persist.sys.timezone')
    device_timezone = value.strip()
    try:
        ZoneInfo(device_timezone)
        result['device_timezone'] = device_timezone
    except (ValueError, ZoneInfoNotFoundError):
        result['device_timezone'] = 'unknown'
    code, value = shell('dumpsys', 'dropbox', 'SYSTEM_LAST_KMSG')
    result['kernel_metadata_dates'] = [m.group(1) for m in ENTRY.finditer(value)]
    if include_kernel:
        if result['primary_user_state'] != 'RUNNING_UNLOCKED':
            result['kernel'] = {'status': 'requires_user_unlock', 'raw_body_saved': False}
        else:
            code, value = shell('dumpsys', 'dropbox', '--print', 'SYSTEM_LAST_KMSG')
            result['kernel'] = kernel_projection(value, result['device_timezone'], after)
            result['kernel']['read_exit_code'] = code
    return result


def write_private_output(path, output):
    # Exclusive creation also rejects existing symlinks. Set the private mode
    # before any bytes are written, rather than chmod after exposing the data.
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, 'O_NOFOLLOW', 0)
    descriptor = os.open(path, flags, 0o600)
    with os.fdopen(descriptor, 'w', encoding='utf-8') as stream:
        stream.write(output)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--adb', default='adb')
    parser.add_argument('--serial', required=True)
    parser.add_argument('--kernel', action='store_true')
    parser.add_argument('--after', help='Earliest relevant UTC ISO timestamp; excludes stale kernel entries')
    parser.add_argument('--output', type=Path, help='New JSON file; refuses to overwrite')
    args = parser.parse_args()
    after = datetime.fromisoformat(args.after.replace('Z', '+00:00')) if args.after else None
    if after is not None and after.tzinfo is None:
        parser.error('--after must include a timezone')
    result = collect(args.adb, args.serial, args.kernel, after)
    output = json.dumps(result, indent=2) + '\n'
    if args.output:
        write_private_output(args.output, output)
    print(output, end='')


if __name__ == '__main__':
    main()