#!/usr/bin/python3
"""
Hardware test for Validity TLS session management (Iteration 2).

Requires a real Validity/Synaptics sensor (06cb:009a or similar) that has
been paired at least once (e.g. via python-validity or Windows driver).

Run with:
  sudo LD_LIBRARY_PATH=builddir/libfprint \
       GI_TYPELIB_PATH=builddir/libfprint \
       FP_DEVICE_EMULATION=0 \
       FP_DRIVERS_ALLOWLIST=validity \
       G_MESSAGES_DEBUG=all \
       python3 tests/validity/test_tls_hardware.py 2>&1

The test will:
  1. Enumerate and detect the validity sensor
  2. Open the device (triggers: GET_VERSION, CMD19, GET_FW_INFO,
     flash read, PSK derivation, flash parse, TLS handshake)
  3. Report whether TLS handshake succeeded or failed
  4. Close the device cleanly
"""

import os
import re
import sys
import traceback

import gi
gi.require_version('FPrint', '2.0')
from gi.repository import FPrint, GLib

# Exit with error on any exception, including in callbacks
sys.excepthook = lambda *args: (traceback.print_exception(*args), sys.exit(1))

# Ensure we're not in emulation mode
if os.environ.get('FP_DEVICE_EMULATION') == '1':
    print('ERROR: FP_DEVICE_EMULATION=1 is set, this test needs real hardware')
    sys.exit(1)

# Ensure running as root (USB access)
if os.geteuid() != 0:
    print('WARNING: Not running as root — USB access may fail')

# Collect debug log lines for analysis
log_lines = []
original_handler = None

def log_handler(log_domain, log_level, message, user_data):
    log_lines.append(message)
    # Also print to stderr for real-time visibility
    print(f'  [{log_domain}] {message}', file=sys.stderr)

# Install log handler to capture libfprint debug output
log_flags = (GLib.LogLevelFlags.LEVEL_DEBUG |
             GLib.LogLevelFlags.LEVEL_INFO |
             GLib.LogLevelFlags.LEVEL_MESSAGE |
             GLib.LogLevelFlags.LEVEL_WARNING |
             GLib.LogLevelFlags.LEVEL_CRITICAL)

for domain in ['libfprint', 'libfprint-SSM', 'libfprint-validity',
               'libfprint-device', 'libfprint-context']:
    GLib.log_set_handler(domain, log_flags, log_handler, None)

print('=== Validity TLS Hardware Test ===')
print()

# Step 1: Enumerate devices
c = FPrint.Context()
c.enumerate()
devices = c.get_devices()

if len(devices) == 0:
    print('FAIL: No fingerprint devices found')
    sys.exit(1)

d = devices[0]
del devices

driver = d.get_driver()
print(f'Found device: driver={driver}')

if driver != 'validity':
    print(f'SKIP: Expected validity driver, got {driver}')
    sys.exit(77)  # meson skip code

# Step 2: Open device (this triggers the full TLS flow)
print()
print('Opening device (GET_VERSION → CMD19 → FW_INFO → Flash Read → PSK → TLS handshake)...')
try:
    d.open_sync()
    print('Device opened successfully')
except GLib.Error as e:
    print(f'FAIL: open_sync() failed: {e.message}')
    sys.exit(1)

# Step 3: Analyze debug log for TLS progress
print()
print('=== TLS Progress Analysis ===')

checks = {
    'fwext_loaded': False,
    'fwext_not_loaded': False,
    'flash_read': False,
    'flash_bytes': None,
    'psk_derived': False,
    'psk_product': None,
    'flash_cert': False,
    'flash_privkey': False,
    'flash_ecdh': False,
    'keys_loaded': False,
    'tls_started': False,
    'server_hello': False,
    'handshake_done': False,
    'secure_session': False,
    'handshake_failed': None,
    'flash_parse_failed': None,
    'no_fwext_skip': False,
}

for line in log_lines:
    if 'Firmware extension is loaded' in line:
        checks['fwext_loaded'] = True

    if 'Firmware extension not loaded' in line:
        checks['fwext_not_loaded'] = True

    if 'No firmware extension' in line:
        checks['no_fwext_skip'] = True
    if 'TLS flash read: got' in line:
        checks['flash_read'] = True
        m = re.search(r'got (\d+) bytes', line)
        if m:
            checks['flash_bytes'] = int(m.group(1))

    if 'PSK derived from DMI' in line:
        checks['psk_derived'] = True
        m = re.search(r'product=(\S+)', line)
        if m:
            checks['psk_product'] = m.group(1)

    if 'TLS flash: certificate loaded' in line:
        checks['flash_cert'] = True

    if 'TLS flash: private key loaded' in line:
        checks['flash_privkey'] = True

    if 'TLS flash: ECDH public key loaded' in line:
        checks['flash_ecdh'] = True

    if 'TLS flash: all keys loaded' in line:
        checks['keys_loaded'] = True

    if 'TLS ServerHello: cipher 0xC005' in line:
        checks['server_hello'] = True

    if 'TLS handshake completed' in line:
        checks['handshake_done'] = True

    if 'TLS session established' in line:
        checks['secure_session'] = True
        checks['tls_started'] = True

    if 'TLS handshake failed' in line:
        checks['handshake_failed'] = line

    if 'TLS flash parse failed' in line:
        checks['flash_parse_failed'] = line

    if 'skipping TLS' in line.lower() or 'continuing without TLS' in line.lower():
        pass  # noted but not a hard failure for this test


# Report results
def report(label, ok, detail=''):
    status = 'PASS' if ok else 'FAIL'
    extra = f' ({detail})' if detail else ''
    print(f'  [{status}] {label}{extra}')

report('Firmware extension loaded',
       checks['fwext_loaded'],
       'NOT loaded' if checks['fwext_not_loaded'] else '')

report('Flash read executed',
       checks['flash_read'],
       f"{checks['flash_bytes']} bytes" if checks['flash_bytes'] else '')

report('PSK derived from DMI',
       checks['psk_derived'],
       checks['psk_product'] or '')

report('Certificate extracted from flash',
       checks['flash_cert'])

report('Private key decrypted from flash',
       checks['flash_privkey'])

report('ECDH public key extracted from flash',
       checks['flash_ecdh'])

report('All TLS keys loaded',
       checks['keys_loaded'])

report('ServerHello received (cipher 0xC005)',
       checks['server_hello'])

report('TLS handshake completed',
       checks['handshake_done'])

report('Secure TLS session established',
       checks['secure_session'])

if checks['handshake_failed']:
    print(f'  [INFO] Handshake failure: {checks["handshake_failed"]}')

if checks['flash_parse_failed']:
    print(f'  [INFO] Flash parse failure: {checks["flash_parse_failed"]}')

# Step 4: Close device
print()
print('Closing device...')
d.close_sync()
print('Device closed successfully')

del d
del c

# Summary
print()
all_ok = (checks['fwext_loaded'] and checks['flash_read'] and
          checks['psk_derived'] and checks['keys_loaded'] and
          checks['handshake_done'] and checks['secure_session'])
fwext_ok_keys_fail = (checks['fwext_loaded'] and checks['flash_read'] and
                      not checks['keys_loaded'])
no_fwext = checks['no_fwext_skip'] or checks['fwext_not_loaded']

if all_ok:
    print('=== RESULT: ALL TLS CHECKS PASSED ===')
    print('TLS session established with real hardware.')
    sys.exit(0)
elif no_fwext:
    print('=== RESULT: FWEXT NOT LOADED ===')
    print('The firmware extension is not loaded on the sensor.')
    print('This is required for flash access and TLS handshake.')
    print()
    print('To resolve, pair the device first with python-validity:')
    print('  sudo validity-sensors-firmware  # download/upload firmware')
    print('  sudo python3 -c "from validitysensor.init import open; open()"')
    print()
    print('The fwext_loaded check verified the driver correctly detects this.')
    sys.exit(0)  # Not a driver bug — this is expected without fwext
elif fwext_ok_keys_fail:
    print('=== RESULT: PARTIAL — Flash readable but keys incomplete ===')
    print('Flash read succeeded but TLS key material is missing or corrupt.')
    print('Re-pairing with python-validity may fix this.')
    sys.exit(1)
else:
    print('=== RESULT: SOME TLS CHECKS FAILED ===')
    sys.exit(1)