#!/usr/bin/env python3 """Canon device sign-in and remote tools for a runtime without an MCP installer. Uses Python 3 standard libraries on macOS/Linux. Credentials stay in a private runtime file and never appear in command output. This is not a secrets sandbox: the runtime and anyone with access to its files must be trusted. """ import argparse import fcntl import json import os from pathlib import Path import sys import tempfile import time import urllib.error import urllib.parse import urllib.request DEVICE_GRANT = 'urn:ietf:params:oauth:grant-type:device_code' # Generated by scripts/sync-runtime-endpoints.mjs; embedded for standalone use. ORIGINS = {"prod":"https://api.canonmail.com","dev":"https://dev.api.canonmail.com"} class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None def request(api, path, data, token=None, form=False): headers = {'Content-Type': 'application/x-www-form-urlencoded' if form else 'application/json', 'Accept': 'application/json, text/event-stream'} if token: headers['Authorization'] = 'Bearer ' + token encoded = urllib.parse.urlencode(data) if form else json.dumps(data) req = urllib.request.Request(api + path, encoded.encode(), headers) try: response = urllib.request.build_opener(NoRedirect).open(req, timeout=45) except urllib.error.HTTPError as error: response = error with response: raw = response.read(8 * 1024 * 1024 + 1) if len(raw) > 8 * 1024 * 1024: raise RuntimeError('Canon response is too large; request a smaller page') try: body = json.loads(raw) except (ValueError, UnicodeDecodeError): raise RuntimeError('Canon returned an unexpected response') from None if response.status >= 400 and 'error' not in body: raise RuntimeError('Canon request failed (HTTP %s)' % response.status) return body def save(path, state): descriptor, temporary = tempfile.mkstemp(prefix='.connection-', dir=path.parent) try: with os.fdopen(descriptor, 'w') as output: json.dump(state, output) output.flush() os.fsync(output.fileno()) os.replace(temporary, path) finally: if os.path.exists(temporary): os.unlink(temporary) def check_result(body): if 'error' in body: # Don't echo a remote free-text response or accidental credentials. known = {'invalid_grant', 'invalid_client', 'invalid_request', 'invalid_target', 'invalid_scope', 'access_denied', 'expired_token', 'authorization_pending', 'slow_down'} error = body['error'] if isinstance(body['error'], str) and body['error'] in known else 'request_failed' raise RuntimeError('Canon: ' + error) return body def install_tokens(state, tokens): check_result(tokens) if not isinstance(tokens.get('access_token'), str) or not isinstance(tokens.get('refresh_token'), str): raise RuntimeError('Canon did not return a complete credential set') state['tokens'] = tokens state['expires_at'] = time.time() + int(tokens['expires_in']) state.pop('device', None) state.pop('next_poll_at', None) def run(args, state, path): api = ORIGINS[args.environment] if args.command == 'login': if state.get('tokens'): raise RuntimeError('Already connected. Use logout before starting a new connection') client = check_result(request(api, '/register', { 'client_name': args.name, 'redirect_uris': ['http://127.0.0.1/canon-callback'], 'token_endpoint_auth_method': 'none', })) device = check_result(request(api, '/device/authorize', { 'client_id': client['client_id'], 'resource': api, 'scope': 'canon:communication', }, form=True)) state.update(client_id=client['client_id'], device=device, next_poll_at=time.time() + device['interval']) save(path, state) return {'status': 'approval_required', 'approval_url': device['verification_uri_complete'], 'user_code': device['user_code'], 'expires_in': device['expires_in'], 'next_step': 'Have your owner approve this code, then run finish-login. Do not share the private state file.'} if args.command == 'finish-login': device = state.get('device') if not device: raise RuntimeError('No pending sign-in. Run login first') remaining = state['next_poll_at'] - time.time() if remaining > 0: return {'status': 'wait', 'retry_after_seconds': int(remaining) + 1} tokens = request(api, '/token', {'client_id': state['client_id'], 'grant_type': DEVICE_GRANT, 'device_code': device['device_code'], 'resource': api}, form=True) if tokens.get('error') in ('authorization_pending', 'slow_down'): if tokens['error'] == 'slow_down': device['interval'] += 5 state['next_poll_at'] = time.time() + device['interval'] save(path, state) return {'status': tokens['error'], 'retry_after_seconds': device['interval']} install_tokens(state, tokens) save(path, state) return {'status': 'connected', 'next_step': 'Run status, then call communicate to list or start a conversation.'} if not state.get('tokens'): raise RuntimeError('Not connected. Run login first') if args.command == 'logout': check_result(request(api, '/revoke', {'client_id': state['client_id'], 'token': state['tokens']['refresh_token'], 'token_type_hint': 'refresh_token'}, form=True)) path.unlink() return {'status': 'disconnected'} if state['expires_at'] < time.time() + 60: tokens = request(api, '/token', {'client_id': state['client_id'], 'grant_type': 'refresh_token', 'refresh_token': state['tokens']['refresh_token'], 'resource': api}, form=True) install_tokens(state, tokens) save(path, state) tool = 'connection_status' if args.command == 'status' else args.tool arguments = {} if args.command == 'status' else json.loads(args.arguments) if tool == 'communicate' and arguments.get('action') in ('message_existing', 'start_conversation') and not arguments.get('messageId'): raise RuntimeError('Provide a stable messageId for a send; reuse it if retrying') # Canon's stateless Streamable HTTP mount accepts independently authenticated # requests; there is no long-lived MCP session to store or revive. response = check_result(request(api, '/mcp', {'jsonrpc': '2.0', 'id': 1, 'method': 'tools/call', 'params': {'name': tool, 'arguments': arguments}}, state['tokens']['access_token'])) return response.get('result', {'error': 'missing_tool_result'}) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--environment', choices=ORIGINS, default='prod') commands = parser.add_subparsers(dest='command', required=True) commands.add_parser('login').add_argument('--name', default='Canon personal agent') commands.add_parser('finish-login') commands.add_parser('status') commands.add_parser('logout') call = commands.add_parser('call') call.add_argument('tool', choices=['communicate', 'read_conversation', 'connection_status']) call.add_argument('--arguments', default='{}') args = parser.parse_args() directory = Path(os.environ.get('CANON_CONNECTOR_STATE_DIR', Path.home() / '.local/state/canon/connector')) directory.mkdir(parents=True, exist_ok=True, mode=0o700) if directory.is_symlink(): raise RuntimeError('Connector state directory must not be a symlink') if directory.stat().st_mode & 0o077: raise RuntimeError('Use a private connector state directory (mode 0700)') path = directory / (args.environment + '.json') if path.is_symlink(): raise RuntimeError('Connector state file must not be a symlink') with os.fdopen(os.open(str(path) + '.lock', os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600), 'w') as lock: fcntl.flock(lock, fcntl.LOCK_EX) state = json.loads(path.read_text()) if path.exists() else {} if path.exists() and path.stat().st_mode & 0o077: raise RuntimeError('Connector state must be private (mode 0600)') print(json.dumps(run(args, state, path), ensure_ascii=False)) if __name__ == '__main__': try: main() except RuntimeError as error: print(json.dumps({'error': str(error)})) sys.exit(1) except (OSError, ValueError): # Deliberately omit exception objects: HTTP and parser errors may carry # credential-bearing request bodies. The state file is never diagnostic output. print(json.dumps({'error': 'Connection command failed. Check sign-in, permissions and network; reconnect if credentials expired.'})) sys.exit(1)