diff --git a/README.md b/README.md index cdfaa036..9a20ba82 100644 --- a/README.md +++ b/README.md @@ -202,8 +202,9 @@ Deploy machine learning models to Synapse devices. ### Prerequisites -1. **Docker** — required for model conversion -2. **QAIRT SDK v2.34** (Qualcomm AI Runtime) — required for model conversion for DSP runtime, not required for CPU runtime +1. **For `.pt` models:** the model-conversion extra, `pip install 'science-synapse[model-convert]'` (PyTorch and ONNX). Not needed for `.onnx` or `.dlc` models. +2. **Docker** — required for quantized (DSP) model conversion +3. **QAIRT SDK v2.34** (Qualcomm AI Runtime) — required for model conversion for DSP runtime, not required for CPU runtime ### Quick Start — Deploy a Float Model (CPU) diff --git a/setup.py b/setup.py index 90a09732..bd004bc1 100644 --- a/setup.py +++ b/setup.py @@ -29,7 +29,7 @@ def stamp_api_version(): setup( name="science-synapse", - version="2.7.7", + version="2.8.0", description="Client library and CLI for the Synapse API", author="Science Team", author_email="team@science.xyz", @@ -63,7 +63,7 @@ def stamp_api_version(): ], extras_require={ "model-convert": [ - "onnx>=1.12.0,<1.16.0", + "onnx==1.16.2", # same as the converter container; first release with Python 3.12 wheels is 1.16.0 "torch", "packaging", "protobuf", diff --git a/synapse-api b/synapse-api index cacec782..f9fc77f6 160000 --- a/synapse-api +++ b/synapse-api @@ -1 +1 @@ -Subproject commit cacec782d383336ce42434716b7b911d3bfda04e +Subproject commit f9fc77f6c3a9b5ccfa6479b48f139e7d540e8843 diff --git a/synapse/_api_version.py b/synapse/_api_version.py index cb64f8d4..445971de 100644 --- a/synapse/_api_version.py +++ b/synapse/_api_version.py @@ -1,2 +1,2 @@ # Generated from synapse-api/VERSION by setup.py. Do not edit. -SYNAPSE_API_VERSION = "2.4.1" +SYNAPSE_API_VERSION = "2.6.0" diff --git a/synapse/cli/__main__.py b/synapse/cli/__main__.py index ccd89f1e..0b4f680b 100755 --- a/synapse/cli/__main__.py +++ b/synapse/cli/__main__.py @@ -10,6 +10,7 @@ from synapse import SYNAPSE_API_VERSION from synapse.cli import ( + auth, apps, deploy_model, discover, @@ -74,6 +75,7 @@ def main(): ) subparsers = parser.add_subparsers(title="Commands") discover.add_commands(subparsers) + auth.add_commands(subparsers) rpc.add_commands(subparsers) streaming.add_commands(subparsers) offline_plot.add_commands(subparsers) diff --git a/synapse/cli/auth.py b/synapse/cli/auth.py new file mode 100644 index 00000000..a318df7b --- /dev/null +++ b/synapse/cli/auth.py @@ -0,0 +1,228 @@ +import getpass +import logging +import socket + +import grpc +from google.protobuf.empty_pb2 import Empty +from rich.console import Console + +import synapse as syn +from synapse.api.auth_pb2 import AuthRequest, RevokeAuthRequest +from synapse.client import auth + +logger = logging.getLogger(__name__) + + +def add_commands(subparsers): + pair_parser = subparsers.add_parser( + "pair", help="Request access to a device, approved on the device screen" + ) + pair_parser.add_argument( + "--label", + help="How this machine identifies itself on the device screen " + "(default: user@hostname)", + default=None, + ) + pair_parser.set_defaults(func=pair) + + unpair_parser = subparsers.add_parser( + "unpair", help="Revoke this computer's access to a device and forget its token" + ) + unpair_parser.add_argument( + "identifier", + nargs="?", + default=None, + help="Serial or stored name of a device to forget, matched against " + "~/.scifi-env. Only forgets the local token, without contacting the " + "device -- use this when the device is gone. If omitted, --uri is used " + "to contact the device, revoke access there, then forget the token.", + ) + unpair_parser.set_defaults(func=unpair) + + +def _default_label() -> str: + try: + return f"{getpass.getuser()}@{socket.gethostname()}" + except Exception: + return "unidentified client" + + +def pair(args): + console = Console() + device = syn.Device(args.uri, args.verbose) + + # Info is open, so this works before pairing. It gives us the serial to key + # the token on and the name to show the user. + try: + info = device.rpc.Info(Empty(), timeout=10.0) + except grpc.RpcError as e: + console.print(f"[bold red]Could not reach the device: {e.details()}") + return + except KeyboardInterrupt: + # Ctrl-C during the 10s Info call, before any code is on screen. Without + # this the user gets a raw traceback; nothing has been requested of the + # device yet, so there is nothing to withdraw. + console.print("\n[yellow]Pairing cancelled. Nothing was saved.") + return + + if not info.serial: + console.print("[bold red]This device did not report a serial number; cannot pair.") + return + + label = args.label or _default_label() + + try: + stream = device.rpc.RequestAuth(AuthRequest(client_label=label)) + + challenge_event = next(stream) + if not challenge_event.HasField("challenge"): + console.print("[bold red]Unexpected response from the device.") + return + code = challenge_event.challenge.code + expires_in = challenge_event.challenge.expires_in_sec + + console.print() + console.print(f" Pairing code: [bold cyan]{code}[/bold cyan]") + console.print() + console.print( + f"Check the screen on [bold]{info.name}[/bold]. If it shows the same code, " + f"tap Allow." + ) + console.print(f"[dim]This request expires in {expires_in} seconds.[/dim]") + console.print() + + with console.status("Waiting for approval on the device", spinner="bouncingBall"): + result_event = next(stream) + + if not result_event.HasField("result"): + console.print("[bold red]Unexpected response from the device.") + return + + result = result_event.result + if not result.granted: + console.print(f"[bold red]Not paired: {result.reason}") + return + + auth.save_token(info.serial, info.name, result.token) + console.print( + f"[bold green]Paired with {info.name} ({info.serial}).[/bold green] " + f"Token saved to {auth.env_file_path()}" + ) + + except grpc.RpcError as e: + if e.code() == grpc.StatusCode.UNIMPLEMENTED: + console.print( + "[bold red]This device's firmware does not support pairing. " + "Update the device firmware and try again." + ) + elif e.code() == grpc.StatusCode.FAILED_PRECONDITION: + console.print(f"[bold red]{e.details()}. Try again in a moment.") + else: + console.print(f"[bold red]Pairing failed: {e.details()}") + except KeyboardInterrupt: + console.print("\n[yellow]Pairing cancelled. Nothing was saved.") + + +def unpair(args): + console = Console() + + # A positional identifier is a request to clean up local state without + # touching the network at all -- this is the "device is gone" case, so it + # takes priority even if --uri was also given. + identifier = getattr(args, "identifier", None) + if identifier: + _unpair_by_identifier(console, identifier) + return + + if not args.uri: + console.print( + "[yellow]Specify a device: `unpair ` to remove a " + "local entry, or `--uri ` to look one up by contacting it." + ) + _print_stored_entries(console) + return + + device = syn.Device(args.uri, args.verbose) + try: + info = device.rpc.Info(Empty(), timeout=10.0) + except grpc.RpcError as e: + console.print(f"[bold red]Could not reach the device: {e.details()}") + console.print( + "[dim]Run `unpair ` to remove a local entry " + "without contacting the device.[/dim]" + ) + return + + token = auth.token_for_serial(info.serial) + if token is None: + console.print(f"[yellow]No stored token for {info.name} ({info.serial}).") + return + + try: + # Revokes whichever token the call carries, i.e. this computer's. + device.rpc.RevokeAuth(RevokeAuthRequest(), timeout=10.0) + console.print(f"[bold green]Revoked this computer's access to {info.name}.") + except grpc.RpcError as e: + if e.code() == grpc.StatusCode.UNAUTHENTICATED: + # Already revoked or expired on the device; only the local copy is left. + pass + elif e.code() == grpc.StatusCode.UNIMPLEMENTED: + console.print( + "[yellow]This device's firmware cannot revoke access remotely. " + "Revoke it on the device under Settings > Authorized Devices." + ) + else: + # Keep the token: without it, this computer can no longer revoke itself. + console.print(f"[bold red]Could not revoke access on the device: {e.details()}") + return + + _remove_and_report(console, info.serial, info.name) + + +def _unpair_by_identifier(console, identifier): + """Forget a locally stored token, matched by serial or name -- no network.""" + tokens = auth.load_tokens() + + if identifier in tokens: + _remove_and_report(console, identifier, tokens[identifier][0], contacted=False) + return + + matches = [serial for serial, (name, _) in tokens.items() if name == identifier] + if len(matches) == 1: + serial = matches[0] + _remove_and_report(console, serial, tokens[serial][0], contacted=False) + return + + if not matches: + console.print(f"[bold red]No stored entry matches '{identifier}'.") + _print_stored_entries(console) + return + + console.print( + f"[bold red]'{identifier}' matches more than one stored device; " + "re-run with the serial, which is unique:" + ) + for serial in matches: + console.print(f" [dim]{tokens[serial][0]} (serial {serial})[/dim]") + + +def _print_stored_entries(console): + tokens = auth.load_tokens() + if not tokens: + console.print("[dim]No devices are currently paired.[/dim]") + return + console.print("[dim]Stored entries:[/dim]") + for serial, (name, _) in sorted(tokens.items()): + console.print(f" [dim]{name} (serial {serial})[/dim]") + + +def _remove_and_report(console, serial, name, contacted=True): + if auth.remove_token(serial): + console.print(f"[bold green]Forgot the token for {name} ({serial}).") + if not contacted: + console.print( + "[dim]The device was not contacted. Use `unpair --uri ` to " + "also revoke access there.[/dim]" + ) + else: + console.print(f"[yellow]No stored token for {name} ({serial}).") diff --git a/synapse/cli/deploy_model.py b/synapse/cli/deploy_model.py index f63bfa38..914902ce 100644 --- a/synapse/cli/deploy_model.py +++ b/synapse/cli/deploy_model.py @@ -1,6 +1,8 @@ """CLI command for deploying models to Synapse devices.""" import argparse + +import grpc import os from typing import Optional @@ -8,11 +10,13 @@ from rich import progress from rich.prompt import Confirm +from synapse import Device import synapse.client.sftp as sftp -from synapse.cli.files import setup_connection +from synapse.cli.files import _uses_sftp +from synapse.client import files as files_client # Constants -DEVICE_MODEL_DIR = "/models" +DEVICE_MODEL_DIR = "models" DEFAULT_SFTP_USER = "scifi-sftp" DEFAULT_ENV_FILE = ".scienv" @@ -44,26 +48,21 @@ def add_commands(subparsers: argparse._SubParsersAction): help="Model name on device (default: filename without extension, e.g., 'my_model')", ) + # Used only with device firmware that predates pairing, which still + # deploys over SFTP with its own password. parser.add_argument( - "--username", - type=str, - default=DEFAULT_SFTP_USER, - help=f"SFTP username (default: {DEFAULT_SFTP_USER})", + "--username", type=str, default=DEFAULT_SFTP_USER, + help=f"SFTP username, for older device firmware only (default: {DEFAULT_SFTP_USER})", ) parser.add_argument( - "--env-file", - "-e", - type=str, - default=DEFAULT_ENV_FILE, - help=f"Password env file (default: {DEFAULT_ENV_FILE})", + "--env-file", "-e", type=str, default=DEFAULT_ENV_FILE, + help=f"File the SFTP password is saved in, for older device firmware only (default: {DEFAULT_ENV_FILE})", ) parser.add_argument( - "--forget-password", - "-f", - action="store_true", - help="Don't store password locally", + "--forget-password", "-f", action="store_true", + help="Do not save the SFTP password, for older device firmware only", ) parser.add_argument( @@ -234,59 +233,125 @@ def deploy_model(args): console.print() - # Step 2: Connect to device via SFTP + # Step 2: Connect to the device. No SFTP password: the pairing token that + # governs every other RPC governs this too. console.print("[bold cyan]Connecting to device...[/bold cyan]") - - result = setup_connection( - args.uri, - args.username, - args.env_file, - args.forget_password, - console, - ) - - if result is None: + device = Device(args.uri, getattr(args, "verbose", False)) + if _uses_sftp(device, console): + if _deploy_over_sftp(args, deploy_path, model_name, remote_ext, console): + _report_deployed(console, model_name, remote_ext, quantize) return - ssh, sftp_conn = result - try: - # Step 3: Ensure model directory exists - _ensure_model_dir(sftp_conn, ssh, console) + # Step 3: No mkdir. WriteFile creates parent directories for whatever + # path the stream names, so models/ appears when the model lands. - # Step 4: Check if model already exists on device + # Step 4: Check if the model already exists. This is only the overwrite + # prompt -- ListFiles rather than a stat, since that is the RPC that + # reports existence. remote_path = f"{DEVICE_MODEL_DIR}/{model_name}{remote_ext}" try: - sftp_conn.stat(remote_path) - if not args.force: - if not Confirm.ask( - f"[yellow]Model '{model_name}{remote_ext}' already exists on device. Overwrite?[/yellow]", - default=False, - ): - console.print("[dim]Aborted.[/dim]") - return - except FileNotFoundError: - pass + existing = {f.path for f in files_client.list_files(device, DEVICE_MODEL_DIR)} + except grpc.RpcError as e: + if e.code() == grpc.StatusCode.UNAUTHENTICATED: + console.print( + "[bold red]Not paired with this device.[/bold red] Run " + "[bold]synapsectl pair[/bold] first." + ) + return + # Any other failure here (a missing models dir, for instance) just + # means we cannot pre-check; the upload itself will still report. + existing = set() + + if remote_path in existing and not args.force: + if not Confirm.ask( + f"[yellow]Model '{model_name}{remote_ext}' already exists on device. Overwrite?[/yellow]", + default=False, + ): + console.print("[dim]Aborted.[/dim]") + return # Step 5: Upload the model file - _upload_file(sftp_conn, deploy_path, remote_path, console) + _upload_file(device, deploy_path, remote_path, console) + _report_deployed(console, model_name, remote_ext, quantize) + finally: + pass + + +def _report_deployed(console: Console, model_name: str, remote_ext: str, quantize: bool): + console.print() + console.print("[bold green]Model deployed successfully![/bold green]") + console.print() + console.print(f" Model deployed: [cyan]models/{model_name}{remote_ext}[/cyan]") + if quantize: + console.print(f" Runtime: [cyan]DSP (quantized INT8)[/cyan]") + else: + console.print(f" Runtime: [cyan]CPU (float32, ONNX Runtime)[/cyan]") console.print() - console.print("[bold green]Model deployed successfully![/bold green]") - console.print() - console.print(f" Model deployed: [cyan]models/{model_name}{remote_ext}[/cyan]") - if quantize: - console.print(f" Runtime: [cyan]DSP (quantized INT8)[/cyan]") - else: - console.print(f" Runtime: [cyan]CPU (float32, ONNX Runtime)[/cyan]") - console.print() - console.print( - " [dim]Tip: for faster DSP inference (~1ms), redeploy with --quantize --input-list[/dim]" - ) - console.print() - console.print(" To load in your app:") - console.print(f' [cyan]auto model = synapse::create_model("{model_name}");[/cyan]') + console.print( + " [dim]Tip: for faster DSP inference (~1ms), redeploy with --quantize --input-list[/dim]" + ) + console.print() + console.print(" To load in your app:") + console.print(f' [cyan]auto model = synapse::create_model("{model_name}");[/cyan]') + +def _upload_file(device, local_path: str, remote_path: str, console: Console): + """Upload a file to the device with progress display.""" + file_size = os.path.getsize(local_path) + + console.print(f"[blue]Uploading to {remote_path}...[/blue]") + + prog = progress.Progress( + progress.SpinnerColumn(), + progress.TextColumn("[progress.description]{task.description}"), + progress.BarColumn(), + progress.DownloadColumn(), + progress.TransferSpeedColumn(), + progress.TimeElapsedColumn(), + ) + + with prog: + task = prog.add_task("Uploading model", total=file_size) + + def update_progress(transferred: int, total: int): + prog.update(task, completed=transferred) + + files_client.write_file( + device, local_path, remote_path, progress=update_progress + ) + + console.print(f"[green]Uploaded to {remote_path}[/green]") + + +# --- Older device firmware: deploy over SFTP, as before the move to gRPC. --- + +# A path inside the scifi-sftp chroot, which is the data directory: the same +# place DEVICE_MODEL_DIR names for the file RPCs. +SFTP_MODEL_DIR = f"/{DEVICE_MODEL_DIR}" + + +def _deploy_over_sftp(args, deploy_path: str, model_name: str, remote_ext: str, console: Console) -> bool: + result = sftp.setup_connection(args.uri, args.username, args.env_file, args.forget_password, console) + if result is None: + return False + ssh, sftp_conn = result + try: + _ensure_model_dir(sftp_conn, ssh, console) + remote_path = f"{SFTP_MODEL_DIR}/{model_name}{remote_ext}" + try: + sftp_conn.stat(remote_path) + if not args.force and not Confirm.ask( + f"[yellow]Model '{model_name}{remote_ext}' already exists on device. Overwrite?[/yellow]", + default=False, + ): + console.print("[dim]Aborted.[/dim]") + return False + except FileNotFoundError: + pass + _sftp_upload_file(sftp_conn, deploy_path, remote_path, console) + return True finally: sftp.close_sftp(ssh, sftp_conn) @@ -298,29 +363,29 @@ def _ensure_model_dir(sftp_conn, ssh, console: Console): falls back to SSH exec to create and chown the directory. """ try: - sftp_conn.stat(DEVICE_MODEL_DIR) + sftp_conn.stat(SFTP_MODEL_DIR) return except FileNotFoundError: pass - console.print(f"[blue]Creating model directory: {DEVICE_MODEL_DIR}[/blue]") + console.print(f"[blue]Creating model directory: {SFTP_MODEL_DIR}[/blue]") # Try SFTP mkdir first try: - sftp_conn.mkdir(DEVICE_MODEL_DIR) + sftp_conn.mkdir(SFTP_MODEL_DIR) return except Exception: pass # SFTP failed — try SSH command to create with proper ownership - real_path = f"/opt/scifi/data{DEVICE_MODEL_DIR}" + real_path = f"/opt/scifi/data{SFTP_MODEL_DIR}" try: _, stdout, stderr = ssh.exec_command( f"mkdir -p {real_path} && chown $(whoami) {real_path}" ) exit_code = stdout.channel.recv_exit_status() if exit_code == 0: - console.print(f"[green]Created {DEVICE_MODEL_DIR}[/green]") + console.print(f"[green]Created {SFTP_MODEL_DIR}[/green]") return except Exception: pass @@ -330,7 +395,7 @@ def _ensure_model_dir(sftp_conn, ssh, console: Console): ) -def _upload_file(sftp_conn, local_path: str, remote_path: str, console: Console): +def _sftp_upload_file(sftp_conn, local_path: str, remote_path: str, console: Console): """Upload a file to the device with progress display.""" file_size = os.path.getsize(local_path) diff --git a/synapse/cli/files.py b/synapse/cli/files.py index ab3844cc..ab4945fb 100644 --- a/synapse/cli/files.py +++ b/synapse/cli/files.py @@ -1,419 +1,432 @@ -import os -import paramiko +"""`synapsectl file` -- device files over gRPC. + +This used SFTP, which needed its own password: a device carried two +independent credentials, and `synapsectl file` was the only command that asked +for one. Going through the SynapseDevice service means file access is governed +by the same pairing token as everything else. + +Device firmware that predates the file RPCs answers them with UNIMPLEMENTED; +for those devices every command falls back to the old SFTP path (files_sftp), +which still takes --username, --env-file and the SFTP password, and says to +update the device firmware. + +Paths are relative to the device's data root and cannot escape it, so +`file ls /` shows the data root. That is the same tree the scifi-sftp account +was chrooted to, so what you can reach is unchanged -- only the credential is. +""" + import argparse -import stat -import yaml -import getpass -from typing import Optional -import paramiko.ssh_exception +import os +from typing import List, Optional + +import grpc +from rich import filesize, progress from rich.console import Console -from rich.table import Table -from rich import progress from rich.prompt import Confirm +from rich.table import Table from synapse import Device -import synapse.client.sftp as sftp -from synapse.utils.file import format_mode, format_time, filesize_binary - -SCIFI_DEFAULT_SFTP_USER = "scifi-sftp" -DEFAULT_ENV_FILE = ".scienv" - +from synapse.api.files_pb2 import ListFilesResponse +from synapse.cli import files_sftp +from synapse.client import files as files_client +from synapse.utils.file import filesize_binary, format_time -def add_user_arguments(parser: argparse.ArgumentParser): +def _add_sftp_flags(parser: argparse.ArgumentParser): + """SFTP settings, used only with device firmware that predates pairing.""" parser.add_argument( - "--username", - "-u", - type=str, - default=SCIFI_DEFAULT_SFTP_USER, - help="Username for SFTP connection", + "--username", "-u", type=str, default=files_sftp.SCIFI_DEFAULT_SFTP_USER, + help="SFTP username, for older device firmware only", ) parser.add_argument( - "--env-file", - "-e", - type=str, - default=DEFAULT_ENV_FILE, - help="Path to environment file containing passwords", + "--env-file", "-e", type=str, default=files_sftp.DEFAULT_ENV_FILE, + help="File the SFTP password is saved in, for older device firmware only", ) parser.add_argument( - "--forget-password", - "-f", - action="store_true", - help="Dont store an input password locally for future use", + "--forget-password", "-f", action="store_true", + help="Do not save the SFTP password, for older device firmware only", ) +def _uses_sftp(device, console: Console) -> bool: + """True for device firmware without the file RPCs, which still serves SFTP. + + Such firmware also predates pairing, so this is where users of a newer + synapsectl find out the device should be updated. + """ + try: + files_client.list_files(device, "") + return False + except grpc.RpcError as e: + if e.code() != grpc.StatusCode.UNIMPLEMENTED: + return False # a real error; let the command report it + console.print( + "[yellow]This device's firmware predates device pairing, so file transfer " + "uses SFTP and needs the device's file-transfer password. Update the device " + "firmware to pair with it instead.[/yellow]" + ) + return True + + def add_commands(subparsers: argparse._SubParsersAction): file_parser = subparsers.add_parser("file", help="File commands") file_subparsers = file_parser.add_subparsers(title="File Commands") - a: argparse.ArgumentParser = file_subparsers.add_parser( - "ls", help="List files on device" + + a = file_subparsers.add_parser("ls", help="List files on device") + a.add_argument( + "path", type=str, nargs="?", default="", help="Path to list, relative to the data directory" + ) + a.add_argument( + "--recursive", "-r", action="store_true", help="List subdirectories too" ) a.add_argument( - "path", type=str, nargs="?", default="/", help="Path to list files from" + "--all", "-a", action="store_true", help="Include hidden files (names starting with .)" ) - add_user_arguments(a) + _add_sftp_flags(a) a.set_defaults(func=ls) - b: argparse.ArgumentParser = file_subparsers.add_parser( - "get", help="Get a file from device" - ) + b = file_subparsers.add_parser("get", help="Get a file from device") b.add_argument("remote_path", type=str, help="Remote path of file to download") b.add_argument( - "--output_path", - "-o", - type=str, - default=os.getcwd(), - help="Output path for downloaded file(s)", + "--output_path", "-o", type=str, default=os.getcwd(), help="Output path for downloaded file(s)" + ) + b.add_argument( + "--recursive", "-r", action="store_true", help="Download directories recursively" ) b.add_argument( - "--recursive", - "-r", + "--no-resume", action="store_true", - help="Download directories recursively", + help="Download from the start instead of continuing an interrupted download", ) - add_user_arguments(b) + b.add_argument( + "--yes", "-y", action="store_true", help="Overwrite existing local files without asking" + ) + _add_sftp_flags(b) b.set_defaults(func=get) - c: argparse.ArgumentParser = file_subparsers.add_parser( - "rm", help="Remove a file from device" + c = file_subparsers.add_parser("put", help="Upload a file to the device") + c.add_argument("local_path", type=str, help="Local file to upload") + c.add_argument( + "remote_path", + type=str, + nargs="?", + default=None, + help="Remote path, relative to the data directory (default: the name as given)", ) - c.add_argument("path", type=str, help="Path to file to remove") c.add_argument( - "--recursive", "-r", action="store_true", help="Remove directories recursively" + "--recursive", "-r", action="store_true", help="Upload a directory recursively" ) - add_user_arguments(c) - c.set_defaults(func=rm) - + c.add_argument( + "--yes", "-y", action="store_true", help="Overwrite existing device files without asking" + ) + _add_sftp_flags(c) + c.set_defaults(func=put) -def ls(args): - console = Console() - connections = setup_connection( - args.uri, - args.username, - args.env_file, - args.forget_password, - console, + d = file_subparsers.add_parser("rm", help="Remove a file from device") + d.add_argument("path", type=str, help="Path to file to remove") + d.add_argument( + "--recursive", "-r", action="store_true", help="Remove directories recursively" ) - if connections is None: - return - ssh, sftp_conn = connections - console.print( - f"\n[bold blue]Listing directory:[/bold blue] [yellow]{args.path}[/yellow]\n" + d.add_argument( + "--yes", "-y", action="store_true", help="Do not ask before deleting" ) + _add_sftp_flags(d) + d.set_defaults(func=rm) - try: - file_attr = sftp_conn.listdir_attr(args.path) - print_file_list(file_attr, console) - except Exception as e: - console.print(f"[bold red]Failed to list directory:[/bold red] {e}") - sftp.close_sftp(ssh, sftp_conn) - - -def get(args): - console = Console() - connections = setup_connection( - args.uri, - args.username, - args.env_file, - args.forget_password, - console, - ) - if connections is None: - return - ssh, sftp_conn = connections - - if args.recursive: - get_dir(sftp_conn, args.remote_path, args.output_path, console) +def _rpc_error(console: Console, action: str, e: grpc.RpcError) -> None: + """One place for gRPC failures, so every command explains them the same way.""" + code = e.code() + if code == grpc.StatusCode.UNAUTHENTICATED: + console.print( + f"[bold red]Not paired with this device.[/bold red] Run " + f"[bold]synapsectl pair[/bold] first." + ) + elif code == grpc.StatusCode.UNIMPLEMENTED: + console.print( + "[bold red]This device's firmware does not support file transfers " + "over gRPC.[/bold red] Update the device firmware and try again." + ) + elif code == grpc.StatusCode.INVALID_ARGUMENT: + console.print(f"[bold red]{e.details()}[/bold red]") + elif code == grpc.StatusCode.NOT_FOUND: + console.print("[bold red]No such file on the device.[/bold red]") else: - get_file(sftp_conn, args.remote_path, args.output_path, console) - - sftp.close_sftp(ssh, sftp_conn) - + console.print(f"[bold red]Failed to {action}:[/bold red] {e.details()}") -def rm(args): - console = Console() - connections = setup_connection( - args.uri, - args.username, - args.env_file, - args.forget_password, - console, - ) - if connections is None: - return - ssh, sftp_conn = connections - - remove_file(sftp_conn, args.path, args.recursive, console) - sftp.close_sftp(ssh, sftp_conn) - - -def setup_connection( - uri: str, - username: str, - env_file: str, - forget_password: bool, - console: Console, -) -> Optional[tuple[paramiko.SSHClient, paramiko.SFTPClient]]: - dev_name = Device(uri).get_name() - password = find_password( - dev_name, env_file - ) # Check if password is provided or stored in env file - if password is None: - console.print(f"[bold red]Didnt find any password for {uri}[/bold red]") - return - # Open SFTP connection - with console.status("Connecting to Synapse device...", spinner="bouncingBall"): - try: - ssh, sftp_conn = sftp.connect_sftp(uri, username, password) - except paramiko.ssh_exception.AuthenticationException: - console.print(f"[bold red]Authentication failed for {uri}[/bold red]") - console.print("[yellow] Incorrect username or password.") - return None - if ssh is None or sftp_conn is None: - console.print(f"[bold red]Failed to connect to {uri}[/bold red]") - return - # If the connection is successful, we can prompt the user if they want to save the password - if not forget_password and dev_name is not None: - save_password(password, env_file, dev_name) - return ssh, sftp_conn - - -def print_file_list(files: list[paramiko.SFTPAttributes], console: Console): - # Sort files: directories first, then by name - files.sort( - key=lambda f: ( - 0 if hasattr(f, "st_mode") and f.st_mode & 0o40000 else 1, - f.filename, - ) - ) +def _print_file_list(files: List[ListFilesResponse.File], console: Console): + # Directories first, then by name -- same ordering the SFTP version used. + entries = sorted(files, key=lambda f: (0 if f.is_dir else 1, f.path)) table = Table(show_header=True) - table.add_column("Permissions", style="cyan") table.add_column("Size", justify="right", style="magenta") table.add_column("Date modified", style="yellow") table.add_column("Filename", style="white") - for file_attr in files: - # Get file mode and format it - mode = format_mode(file_attr.st_mode if hasattr(file_attr, "st_mode") else None) - - # Filesize - size = file_attr.st_size if hasattr(file_attr, "st_size") else 0 - size_str = filesize_binary(size) - - # Time modified - mtime = file_attr.st_mtime if hasattr(file_attr, "st_mtime") else None - time_str = format_time(mtime) - - # Determine filename style based on type - filename = file_attr.filename - is_dir = stat.S_ISDIR(file_attr.st_mode) - is_link = stat.S_ISLNK(file_attr.st_mode) - is_executable = hasattr(file_attr, "st_mode") and file_attr.st_mode & 0o100 - - if is_dir: - filename_styled = f"[bold blue]{filename}/[/bold blue]" - elif is_link: - filename_styled = f"[cyan]{filename}@[/cyan]" - elif is_executable: - filename_styled = f"[bold green]{filename}*[/bold green]" - else: - filename_styled = filename - - table.add_row(mode, size_str, time_str, filename_styled) + for f in entries: + size_str = "" if f.is_dir else filesize_binary(f.size_bytes) + name = f"[bold blue]{f.path}/[/bold blue]" if f.is_dir else f.path + table.add_row(size_str, format_time(f.modified), name) console.print(table) + if not entries: + console.print("[dim](empty)[/dim]") -def get_dir( - sftp_conn: paramiko.SFTPClient, remote_path: str, output_path: str, console: Console -): - try: - dir_attrs = sftp_conn.stat(remote_path) - except FileNotFoundError: - console.print( - f"[bold red]Directory not found:[/bold red] [bold blue]{remote_path}" - ) - return - if not stat.S_ISDIR(dir_attrs.st_mode): - get_file(sftp_conn, remote_path, output_path, console) - return - - output_path = os.path.join(output_path, os.path.basename(remote_path)) - local_dir = os.path.dirname(output_path) - try: - if local_dir != "": - os.makedirs(local_dir, exist_ok=True) - except OSError: - console.print( - f"[bold red]Failed to create output directory:[/bold red] {local_dir}" - ) - return - - file_list = sftp_conn.listdir(remote_path) +def _is_hidden(path: str, listed: str) -> bool: + """Whether `path` is, or sits inside, a dotfile below the listed directory. - for file in file_list: - remote_file = os.path.join(remote_path, file) - local_file = os.path.join(output_path, file) - if stat.S_ISDIR(sftp_conn.stat(remote_file).st_mode): - get_dir(sftp_conn, remote_file, local_file, console) - else: - get_file(sftp_conn, remote_file, local_file, console) + Judged only below `listed`, so explicitly listing a hidden directory still + shows what is in it. + """ + prefix = listed.strip("/") + rel = path.strip("/") + if prefix and (rel == prefix or rel.startswith(prefix + "/")): + rel = rel[len(prefix):].lstrip("/") + return any(part.startswith(".") for part in rel.split("/") if part) -def get_file( - sftp_conn: paramiko.SFTPClient, remote_path: str, output_path: str, console: Console -): +def ls(args): + console = Console() + device = Device(args.uri, args.verbose) + if _uses_sftp(device, console): + return files_sftp.ls(args) + shown = args.path or "/" + console.print(f"\n[bold blue]Listing directory:[/bold blue] [yellow]{shown}[/yellow]\n") try: - is_dir = stat.S_ISDIR(sftp_conn.stat(remote_path).st_mode) - except FileNotFoundError: - console.print(f"[bold red]File not found:[/bold red] [blue]{remote_path}") + entries = files_client.list_files(device, args.path, args.recursive) + except grpc.RpcError as e: + _rpc_error(console, "list the directory", e) return + if not getattr(args, "all", False): + entries = [f for f in entries if not _is_hidden(f.path, args.path)] + _print_file_list(entries, console) - if is_dir: - console.print( - f"[bold yellow]Requested path: [/bold yellow][blue]{remote_path}[/blue][bold yellow] is a directory. Use the --recursive(-r) flag to download directories." - ) - return - local_dir = os.path.dirname(output_path) +def _may_overwrite(console: Console, what: str) -> bool: + """Ask before replacing `what`. Declines when there is no terminal to ask on.""" try: - if local_dir != "": - os.makedirs(local_dir, exist_ok=True) - except OSError: + ok = Confirm.ask(f"[yellow]{what}[/yellow] already exists. Overwrite?", default=False) + except EOFError: + ok = False + if not ok: + console.print(f"Skipped {what} (pass -y to overwrite)") + return ok + + +def _download_one( + device, console: Console, remote: str, local: str, resume: bool, overwrite: bool, + stamp: Optional[files_client.Stamp] = None, +) -> Optional[bool]: + """True if downloaded, False if it failed, None if the user chose to skip it.""" + if os.path.exists(local) and not overwrite and not _may_overwrite(console, local): + return None + with progress.Progress( + progress.TextColumn("[cyan]{task.description}"), + progress.BarColumn(), + progress.DownloadColumn(), + progress.TransferSpeedColumn(), + progress.TimeRemainingColumn(), + console=console, + ) as bar: + task = bar.add_task(os.path.basename(remote), total=None) + + def on_progress(done: int, total: int): + # The device reports the total on every chunk, so it is only known + # once the first one lands. + bar.update(task, completed=done, total=total or None) + + try: + files_client.read_file( + device, remote, local, progress=on_progress, resume=resume, stamp=stamp + ) + return True + except grpc.RpcError as e: + _rpc_error(console, f"download {remote}", e) + _report_kept_partial(console, local) + return False + except KeyboardInterrupt: + _report_kept_partial(console, local) + raise SystemExit(130) + + +def _report_kept_partial(console: Console, local: str) -> None: + """Say where an interrupted download went, so it is not a mystery file.""" + partial = local + files_client.PARTIAL_SUFFIX + if os.path.exists(partial): console.print( - f"[bold red]Failed to create output directory:[/bold red] {local_dir}" + f"Kept {filesize.decimal(os.path.getsize(partial))} in [cyan]{partial}[/cyan]. " + "Run the same command to resume it, or add --no-resume to start over." ) - return - local_path = output_path - if os.path.isdir(output_path): - local_path = os.path.join(output_path, os.path.basename(remote_path)) - try: - file_st_size = sftp_conn.stat(remote_path).st_size - file_size = file_st_size if file_st_size is not None else 0 - - prog = progress.Progress( - progress.SpinnerColumn(), - progress.TextColumn("[progress.description]{task.description}"), - progress.BarColumn(), - progress.DownloadColumn(), - progress.TransferSpeedColumn(), - progress.TimeElapsedColumn(), +def get(args): + console = Console() + device = Device(args.uri, args.verbose) + if _uses_sftp(device, console): + return files_sftp.get(args) + resume = not args.no_resume + + if not args.recursive: + # Checked first so a typo gets one plain line, not an overwrite prompt + # and an empty progress bar. + entry = _remote_entry(device, args.remote_path) + if entry is None: + console.print(f"[bold red]{args.remote_path} does not exist on the device.[/bold red]") + raise SystemExit(1) + if entry.is_dir: + console.print( + f"[bold red]{args.remote_path} is a directory.[/bold red] Pass --recursive to download it." + ) + raise SystemExit(1) + local = args.output_path + if os.path.isdir(local): + local = os.path.join(local, os.path.basename(args.remote_path)) + result = _download_one( + device, console, args.remote_path, local, resume, args.yes, files_client.stamp_of(entry) ) - with prog: - task = prog.add_task(f"Downloading file: {remote_path}", total=file_size) - - def update_progress(transferred: int, total: int): - prog.update(task, completed=transferred) - - sftp_conn.get(remote_path, local_path, callback=update_progress) - except paramiko.SFTPError as e: - console.print(f"[bold red]Failed to download file:[/bold red] {e}") + if result is False: + raise SystemExit(1) + if result: + console.print(f"[bold green]Saved[/bold green] {local}") return - -def remove_file( - sftp_conn: paramiko.SFTPClient, remote_path: str, recursive: bool, console: Console -): try: - is_dir = stat.S_ISDIR(sftp_conn.stat(remote_path).st_mode) - except FileNotFoundError: - console.print(f"[bold red]File not found:[/bold red] [blue]{remote_path}") + entries = files_client.list_files(device, args.remote_path, recursive=True) + except grpc.RpcError as e: + _rpc_error(console, "list the directory", e) return - if is_dir and not recursive: - console.print( - f"[bold yellow]Requested path: [/bold yellow][blue]{remote_path}[/blue][bold yellow] is a directory. Use the --recursive(-r) flag to remove directories." - ) + wanted = [f for f in entries if not f.is_dir] + if not wanted: + console.print("[yellow]Nothing to download.[/yellow]") return + ok = 0 + for f in wanted: + # Mirror the device's layout under the output directory. relpath keeps + # the tree shape without embedding the parent's name twice. + rel = os.path.relpath(f.path, args.remote_path) if args.remote_path else f.path + local = os.path.join(args.output_path, os.path.basename(args.remote_path.rstrip("/")) or "", rel) + if _download_one(device, console, f.path, local, resume, args.yes, files_client.stamp_of(f)): + ok += 1 + console.print(f"[bold green]Downloaded {ok}/{len(wanted)} file(s)[/bold green] to {args.output_path}") + + +def _remote_entry(device, path: str): + """The device's listing entry for `path`, or None if nothing is there. + + Lists the parent and looks for the entry, rather than listing `path` + itself: an empty directory and a missing path both list as nothing, so + listing the target cannot tell them apart. + """ + cleaned = path.rstrip("/") try: - if is_dir: - file_list = sftp_conn.listdir(remote_path) - for file in file_list: - remote_file = os.path.join(remote_path, file) - remove_file(sftp_conn, remote_file, recursive, console) - sftp_conn.rmdir(remote_path) - else: - with console.status( - f"Removing file: {remote_path}", spinner="bouncingBall" - ): - sftp_conn.remove(remote_path) - except Exception as e: - console.print(f"[bold red]Failed to remove file:[/bold red] {e}") - return - - console.print(f"[bold green]File removed:[/bold green] [blue]{remote_path}") + entries = files_client.list_files(device, os.path.dirname(cleaned)) + except grpc.RpcError: + return None + name = os.path.basename(cleaned) + return next((f for f in entries if os.path.basename(f.path) == name), None) -def find_password(dev_name: Optional[str], env_file: Optional[str]): - password = None - if env_file is not None and dev_name is not None: - if os.path.exists(env_file): - password = load_pass_from_env_file(env_file, dev_name) - if password is not None: - return password - password = getpass.getpass("Enter password: ") - return password +def _remote_files_under(device, path: str) -> set: + """Paths of the files already under `path` on the device (empty if none).""" + try: + return {f.path for f in files_client.list_files(device, path, recursive=True) if not f.is_dir} + except grpc.RpcError: + return set() + + +def _upload_one(device, console: Console, local: str, remote: str) -> Optional[int]: + with progress.Progress( + progress.TextColumn("[cyan]{task.description}"), + progress.BarColumn(), + progress.DownloadColumn(), + progress.TransferSpeedColumn(), + console=console, + ) as bar: + task = bar.add_task(os.path.basename(local), total=os.path.getsize(local)) + try: + return files_client.write_file( + device, + local, + remote, + progress=lambda done, tot: bar.update(task, completed=done), + ) + except grpc.RpcError as e: + _rpc_error(console, f"upload {local}", e) + return None -def save_password(password: str, env_file: str, device_name: str): - if env_file is None or device_name is None or password is None: - return - if os.path.exists(env_file): - if load_pass_from_env_file(env_file, device_name) is not None: +def put(args): + console = Console() + device = Device(args.uri, args.verbose) + if _uses_sftp(device, console): + return files_sftp.put(args) + local = args.local_path.rstrip("/") + + if os.path.isdir(local): + if not args.recursive: + console.print( + f"[bold red]{local} is a directory.[/bold red] Pass --recursive to upload it." + ) return + base = args.remote_path or os.path.basename(local) + # No mkdir needed: the server creates parent directories for whatever + # path a chunk stream names, so the tree appears as the files land. + files = [ + os.path.join(root, name) + for root, _, names in os.walk(local) + for name in names + ] + if not files: + console.print("[yellow]Nothing to upload.[/yellow]") + return + # One recursive listing up front rather than one per file. + existing = _remote_files_under(device, base) + ok = 0 + for f in sorted(files): + remote = os.path.join(base, os.path.relpath(f, local)) + if remote in existing and not args.yes and not _may_overwrite(console, remote): + continue + if _upload_one(device, console, f, remote) is not None: + ok += 1 + console.print(f"[bold green]Uploaded {ok}/{len(files)} file(s)[/bold green] to {base}") + return - save_pass = Confirm.ask(f"Save password for {device_name} in {env_file}?") - if not save_pass: + if not os.path.isfile(local): + console.print(f"[bold red]No such local file:[/bold red] {local}") return - store_pass_to_env_file(env_file, device_name, password) + remote = args.remote_path or os.path.basename(local) + # `put file some/dir` means into that directory, the same way + # `get file -o some/dir` does. Without this the server is asked to replace + # a directory with a file, which cannot work. + entry = _remote_entry(device, remote) if remote.strip("/") else None + if not remote.strip("/") or (entry and entry.is_dir): + remote = os.path.join(remote.rstrip("/"), os.path.basename(local)) + entry = _remote_entry(device, remote) + if entry and not args.yes and not _may_overwrite(console, remote): + return + written = _upload_one(device, console, local, remote) + if written is not None: + console.print(f"[bold green]Uploaded[/bold green] {written} bytes to {remote}") -def load_pass_from_env_file(env_file: str, device_name: str) -> Optional[str]: - try: - with open(env_file, "r") as f: - env_loaded = yaml.safe_load(f) - if env_loaded is None: - return None - if "sftp_passwords" not in env_loaded: - return None - passwords = env_loaded.get("sftp_passwords", {}) - if device_name in passwords: - try: - stored_pass = passwords[device_name] - return stored_pass - except Exception: - print( - f"Couldnt read password for {device_name} from env file. Env file may be improperly formatted." - ) - return None - else: - return None - except Exception as e: - print(f"Couldnt read env file at: {env_file}. {e}") - return None - - -# Store password to the .env file in yaml format -def store_pass_to_env_file(env_file: str, device_name: str, password: str): +def rm(args): + console = Console() + # Recursive deletes take a whole recording directory with them, and there + # is no undo on the device. Asked before the SFTP hand-off so both paths ask. + if args.recursive and not args.yes: + if not Confirm.ask(f"Delete [bold]{args.path}[/bold] and everything under it?"): + console.print("Not deleting.") + return + + device = Device(args.uri, args.verbose) + if _uses_sftp(device, console): + return files_sftp.rm(args) + try: - prev_env = {} - if os.path.exists(env_file): - with open(env_file, "r") as f: - prev_env = yaml.safe_load(f) - prev_env = prev_env if prev_env is not None else {} - passwords = prev_env.get("sftp_passwords", {}) - passwords[device_name] = password - prev_env["sftp_passwords"] = passwords - with open(env_file, "w", encoding="utf8") as f: - yaml.dump(prev_env, f, default_flow_style=False) - except TypeError: - print( - f"Failed to store pass to env file at: {env_file}. Env file may be improperly formatted." - ) - except Exception as e: - print(f"Failed to store pass to env file at: {env_file}. {e}") + files_client.delete_file(device, args.path, args.recursive) + except grpc.RpcError as e: + _rpc_error(console, f"delete {args.path}", e) + return + console.print(f"[bold green]Deleted[/bold green] {args.path}") diff --git a/synapse/cli/files_sftp.py b/synapse/cli/files_sftp.py new file mode 100644 index 00000000..de32f209 --- /dev/null +++ b/synapse/cli/files_sftp.py @@ -0,0 +1,364 @@ +"""synapsectl file over SFTP, for device firmware that predates pairing. + +Older firmware has no file RPCs, only the scifi-sftp account and its own +password. This is the SFTP implementation synapsectl shipped before the move +to gRPC, kept so a newer synapsectl still works with those devices; files.py +calls into it when the device answers the file RPCs with UNIMPLEMENTED. +""" + +import os +import stat + +import paramiko +from rich import progress +from rich.console import Console +from rich.table import Table + +import synapse.client.sftp as sftp +from synapse.utils.file import filesize_binary, format_mode, format_time + +SCIFI_DEFAULT_SFTP_USER = "scifi-sftp" +DEFAULT_ENV_FILE = ".scienv" + + +def ls(args): + console = Console() + connections = sftp.setup_connection( + args.uri, + args.username, + args.env_file, + args.forget_password, + console, + ) + if connections is None: + return + ssh, sftp_conn = connections + # The shared ls parser defaults to "" (the gRPC data root); SFTP needs "/". + path = args.path or "/" + console.print( + f"\n[bold blue]Listing directory:[/bold blue] [yellow]{path}[/yellow]\n" + ) + + try: + file_attr = sftp_conn.listdir_attr(path) + print_file_list(file_attr, console) + except Exception as e: + console.print(f"[bold red]Failed to list directory:[/bold red] {e}") + + sftp.close_sftp(ssh, sftp_conn) + + +def get(args): + console = Console() + connections = sftp.setup_connection( + args.uri, + args.username, + args.env_file, + args.forget_password, + console, + ) + if connections is None: + return + ssh, sftp_conn = connections + + if args.recursive: + get_dir(sftp_conn, args.remote_path, args.output_path, console, args.yes) + else: + get_file(sftp_conn, args.remote_path, args.output_path, console, args.yes) + + sftp.close_sftp(ssh, sftp_conn) + + +def rm(args): + console = Console() + connections = sftp.setup_connection( + args.uri, + args.username, + args.env_file, + args.forget_password, + console, + ) + if connections is None: + return + ssh, sftp_conn = connections + + remove_file(sftp_conn, args.path, args.recursive, console) + sftp.close_sftp(ssh, sftp_conn) + + +def print_file_list(files: list[paramiko.SFTPAttributes], console: Console): + # Sort files: directories first, then by name + files.sort( + key=lambda f: ( + 0 if hasattr(f, "st_mode") and f.st_mode & 0o40000 else 1, + f.filename, + ) + ) + + table = Table(show_header=True) + table.add_column("Permissions", style="cyan") + table.add_column("Size", justify="right", style="magenta") + table.add_column("Date modified", style="yellow") + table.add_column("Filename", style="white") + for file_attr in files: + # Get file mode and format it + mode = format_mode(file_attr.st_mode if hasattr(file_attr, "st_mode") else None) + + # Filesize + size = file_attr.st_size if hasattr(file_attr, "st_size") else 0 + size_str = filesize_binary(size) + + # Time modified + mtime = file_attr.st_mtime if hasattr(file_attr, "st_mtime") else None + time_str = format_time(mtime) + + # Determine filename style based on type + filename = file_attr.filename + is_dir = stat.S_ISDIR(file_attr.st_mode) + is_link = stat.S_ISLNK(file_attr.st_mode) + is_executable = hasattr(file_attr, "st_mode") and file_attr.st_mode & 0o100 + + if is_dir: + filename_styled = f"[bold blue]{filename}/[/bold blue]" + elif is_link: + filename_styled = f"[cyan]{filename}@[/cyan]" + elif is_executable: + filename_styled = f"[bold green]{filename}*[/bold green]" + else: + filename_styled = filename + + table.add_row(mode, size_str, time_str, filename_styled) + + console.print(table) + + +def get_dir( + sftp_conn: paramiko.SFTPClient, + remote_path: str, + output_path: str, + console: Console, + overwrite: bool = False, +): + try: + dir_attrs = sftp_conn.stat(remote_path) + except FileNotFoundError: + console.print( + f"[bold red]Directory not found:[/bold red] [bold blue]{remote_path}" + ) + return + if not stat.S_ISDIR(dir_attrs.st_mode): + get_file(sftp_conn, remote_path, output_path, console, overwrite) + return + + output_path = os.path.join(output_path, os.path.basename(remote_path)) + local_dir = os.path.dirname(output_path) + try: + if local_dir != "": + os.makedirs(local_dir, exist_ok=True) + except OSError: + console.print( + f"[bold red]Failed to create output directory:[/bold red] {local_dir}" + ) + return + + file_list = sftp_conn.listdir(remote_path) + + for file in file_list: + remote_file = os.path.join(remote_path, file) + local_file = os.path.join(output_path, file) + if stat.S_ISDIR(sftp_conn.stat(remote_file).st_mode): + get_dir(sftp_conn, remote_file, local_file, console, overwrite) + else: + get_file(sftp_conn, remote_file, local_file, console, overwrite) + + +def get_file( + sftp_conn: paramiko.SFTPClient, + remote_path: str, + output_path: str, + console: Console, + overwrite: bool = False, +): + try: + is_dir = stat.S_ISDIR(sftp_conn.stat(remote_path).st_mode) + except FileNotFoundError: + console.print(f"[bold red]File not found:[/bold red] [blue]{remote_path}") + return + + if is_dir: + console.print( + f"[bold yellow]Requested path: [/bold yellow][blue]{remote_path}[/blue][bold yellow] is a directory. Use the --recursive(-r) flag to download directories." + ) + return + + local_dir = os.path.dirname(output_path) + try: + if local_dir != "": + os.makedirs(local_dir, exist_ok=True) + except OSError: + console.print( + f"[bold red]Failed to create output directory:[/bold red] {local_dir}" + ) + return + + local_path = output_path + if os.path.isdir(output_path): + local_path = os.path.join(output_path, os.path.basename(remote_path)) + if os.path.exists(local_path) and not overwrite and not _may_overwrite(console, local_path): + return + + try: + file_st_size = sftp_conn.stat(remote_path).st_size + file_size = file_st_size if file_st_size is not None else 0 + + prog = progress.Progress( + progress.SpinnerColumn(), + progress.TextColumn("[progress.description]{task.description}"), + progress.BarColumn(), + progress.DownloadColumn(), + progress.TransferSpeedColumn(), + progress.TimeElapsedColumn(), + ) + with prog: + task = prog.add_task(os.path.basename(remote_path), total=file_size) + + def update_progress(transferred: int, total: int): + prog.update(task, completed=transferred) + + sftp_conn.get(remote_path, local_path, callback=update_progress) + except paramiko.SFTPError as e: + console.print(f"[bold red]Failed to download file:[/bold red] {e}") + return + + +def remove_file( + sftp_conn: paramiko.SFTPClient, remote_path: str, recursive: bool, console: Console +): + try: + is_dir = stat.S_ISDIR(sftp_conn.stat(remote_path).st_mode) + except FileNotFoundError: + console.print(f"[bold red]File not found:[/bold red] [blue]{remote_path}") + return + + if is_dir and not recursive: + console.print( + f"[bold yellow]Requested path: [/bold yellow][blue]{remote_path}[/blue][bold yellow] is a directory. Use the --recursive(-r) flag to remove directories." + ) + return + + try: + if is_dir: + file_list = sftp_conn.listdir(remote_path) + for file in file_list: + remote_file = os.path.join(remote_path, file) + remove_file(sftp_conn, remote_file, recursive, console) + sftp_conn.rmdir(remote_path) + else: + with console.status( + f"Removing file: {remote_path}", spinner="bouncingBall" + ): + sftp_conn.remove(remote_path) + except Exception as e: + console.print(f"[bold red]Failed to remove file:[/bold red] {e}") + return + + console.print(f"[bold green]File removed:[/bold green] [blue]{remote_path}") + + +def put(args): + console = Console() + connections = sftp.setup_connection( + args.uri, + args.username, + args.env_file, + args.forget_password, + console, + ) + if connections is None: + return + ssh, sftp_conn = connections + + local = args.local_path.rstrip("/") + remote = args.remote_path or os.path.basename(local) + try: + if os.path.isdir(local): + if not args.recursive: + console.print(f"[bold red]{local} is a directory.[/bold red] Pass --recursive to upload it.") + return + for root, _, names in os.walk(local): + for name in sorted(names): + path = os.path.join(root, name) + put_file(sftp_conn, path, os.path.join(remote, os.path.relpath(path, local)), console, args.yes) + elif os.path.isfile(local): + try: + if stat.S_ISDIR(sftp_conn.stat(remote).st_mode): + remote = os.path.join(remote, os.path.basename(local)) + except FileNotFoundError: + pass + put_file(sftp_conn, local, remote, console, args.yes) + else: + console.print(f"[bold red]No such local file:[/bold red] {local}") + finally: + sftp.close_sftp(ssh, sftp_conn) + + +def _remote_exists(sftp_conn: paramiko.SFTPClient, path: str) -> bool: + try: + sftp_conn.stat(path) + return True + except FileNotFoundError: + return False + + +def _may_overwrite(console: Console, what: str) -> bool: + # Imported here: synapse.cli.files imports this module. + from synapse.cli.files import _may_overwrite as ask + + return ask(console, what) + + +def _makedirs(sftp_conn: paramiko.SFTPClient, remote_dir: str): + """mkdir -p over SFTP.""" + path = "" + for part in [p for p in remote_dir.split("/") if p]: + path = f"{path}/{part}" if path else (f"/{part}" if remote_dir.startswith("/") else part) + try: + sftp_conn.stat(path) + except FileNotFoundError: + sftp_conn.mkdir(path) + + +def put_file( + sftp_conn: paramiko.SFTPClient, + local_path: str, + remote_path: str, + console: Console, + overwrite: bool = False, +): + if not overwrite and _remote_exists(sftp_conn, remote_path) and not _may_overwrite(console, remote_path): + return + try: + remote_dir = os.path.dirname(remote_path) + if remote_dir: + _makedirs(sftp_conn, remote_dir) + prog = progress.Progress( + progress.SpinnerColumn(), + progress.TextColumn("[progress.description]{task.description}"), + progress.BarColumn(), + progress.DownloadColumn(), + progress.TransferSpeedColumn(), + progress.TimeElapsedColumn(), + ) + with prog: + task = prog.add_task(os.path.basename(local_path), total=os.path.getsize(local_path)) + sftp_conn.put(local_path, remote_path, callback=lambda done, _total: prog.update(task, completed=done)) + console.print(f"[bold green]Uploaded[/bold green] {local_path} to {remote_path}") + except PermissionError: + # The SFTP account is chrooted to the data directory, whose top level + # it cannot write; only some folders under it are writable. + console.print( + f"[bold red]The device does not allow uploads to {remote_path}.[/bold red] " + "On this firmware, upload into a folder such as disk_writer/ or models/." + ) + except (OSError, paramiko.SFTPError) as e: + console.print(f"[bold red]Failed to upload {local_path}:[/bold red] {e}") diff --git a/synapse/client/auth.py b/synapse/client/auth.py new file mode 100644 index 00000000..e5078a0d --- /dev/null +++ b/synapse/client/auth.py @@ -0,0 +1,243 @@ +"""Pairing tokens for SciFi devices. + +A token is stored per device serial in a user-wide file (default +``$HOME/.scifi-env``, overridable with ``SCIFI_ENV_FILE``): + + # serial name token + NYX1512-0042 sci-fi-1234 f3a9c1... + +Keyed on serial rather than name because device names are neither unique nor +stable, and keying on a name means renaming a device silently orphans its token. +The name is kept as a human-readable hint only. + +The file is not shell-sourceable: device names contain hyphens, so +``sci-fi-1234=...`` is not a valid shell variable, and mangling names into +``SCIFI_TOKEN_SCI_FI_1234`` risks two names colliding on one key. +""" + +import logging +import os +import tempfile +import threading +from pathlib import Path +from typing import Dict, Optional, Tuple + +import grpc + +logger = logging.getLogger(__name__) + +_HEADER = "# serial\tname\ttoken\n" + + +def env_file_path() -> Path: + override = os.environ.get("SCIFI_ENV_FILE") + if override: + return Path(override) + return Path.home() / ".scifi-env" + + +def load_tokens() -> Dict[str, Tuple[str, str]]: + """Return {serial: (name, token)}. A missing or unreadable file is empty.""" + path = env_file_path() + if not path.exists(): + return {} + + tokens: Dict[str, Tuple[str, str]] = {} + try: + for line in path.read_text().splitlines(): + stripped = line.strip() + if not stripped or stripped.startswith("#"): + continue + fields = stripped.split() + if len(fields) < 3: + logger.debug("Skipping malformed line in %s: %r", path, line) + continue + serial, name, token = fields[0], fields[1], fields[2] + tokens[serial] = (name, token) + except OSError as e: + logger.debug("Could not read %s: %s", path, e) + return {} + return tokens + + +def token_for_serial(serial: str) -> Optional[str]: + entry = load_tokens().get(serial) + return entry[1] if entry else None + + +def has_any_tokens() -> bool: + return bool(load_tokens()) + + +def save_token(serial: str, name: str, token: str) -> None: + """Add or replace the entry for `serial`, leaving other devices untouched.""" + tokens = load_tokens() + tokens[serial] = (name, token) + _write(tokens) + + +def remove_token(serial: str) -> bool: + """Drop the local entry. Returns False if there was nothing to drop.""" + tokens = load_tokens() + if serial not in tokens: + return False + del tokens[serial] + _write(tokens) + return True + + +def _write(tokens: Dict[str, Tuple[str, str]]) -> None: + path = env_file_path() + path.parent.mkdir(parents=True, exist_ok=True) + + lines = [_HEADER] + for serial, (name, token) in sorted(tokens.items()): + lines.append(f"{serial}\t{name}\t{token}\n") + + # Write to a temp file in the same directory, then atomically swap it in + # with os.replace(). A truncate-in-place rewrite can leave the file empty + # or torn if the process dies mid-write (OOM kill, power loss, disk full), + # losing every paired device's token, and offers no protection against two + # concurrent writers interleaving bytes. os.replace() is atomic within a + # filesystem, so a reader always sees either the complete old file or the + # complete new one -- hence the temp file must live next to the target + # (not /tmp, which may be a different mount than $HOME). + # + # mkstemp() creates the file with mode 0600 already (it does not consult + # umask), but we chmod explicitly rather than rely on that, matching how + # the final file is created 0600 from the outset, never briefly + # world-readable, not even under the temp name. + fd, tmp_path = tempfile.mkstemp( + prefix=".scifi-env.", suffix=".tmp", dir=str(path.parent) + ) + try: + os.chmod(tmp_path, 0o600) + with os.fdopen(fd, "w") as f: + f.writelines(lines) + f.flush() + # fsync before the replace: without it, the rename can be durable + # while the file's contents are not, so a crash right after could + # swap in a file that reads back empty. + os.fsync(f.fileno()) + os.replace(tmp_path, path) + except BaseException: + try: + os.remove(tmp_path) + except OSError: + pass + raise + + +# The standard bearer-token header (RFC 6750) rather than a bespoke x- key, so +# proxies and tooling that already understand it keep working. The value carries +# the "Bearer " scheme; the server strips it case-insensitively. +METADATA_KEY = "authorization" +METADATA_SCHEME = "Bearer " + +# Mirrors scifi-server's open set (src/auth/rpc_auth_policy.cpp). These need no +# token, so the interceptor never triggers serial resolution for them. +OPEN_METHODS = frozenset( + { + "/synapse.SynapseDevice/Info", + "/synapse.SynapseDevice/GetLogs", + "/synapse.SynapseDevice/TailLogs", + "/synapse.SynapseDevice/RequestAuth", + } +) + + +def resolve_serial_via_info(channel) -> Optional[str]: + """One Info call to learn the device serial. Info is open, so no token.""" + from google.protobuf.empty_pb2 import Empty + + from synapse.api.synapse_pb2_grpc import SynapseDeviceStub + + try: + info = SynapseDeviceStub(channel).Info(Empty(), timeout=5.0) + return info.serial or None + except grpc.RpcError as e: + logger.debug("Could not read the device serial: %s", e) + return None + + +class AuthInterceptor( + grpc.UnaryUnaryClientInterceptor, + grpc.UnaryStreamClientInterceptor, + grpc.StreamUnaryClientInterceptor, + grpc.StreamStreamClientInterceptor, +): + """Attaches the pairing token to calls that need one. + + Resolution is lazy and happens at most once per channel: the first call to a + token-gated method learns the device serial, looks the token up, and caches + it. Open methods never trigger it, so discovery and log tailing cost nothing + extra, and neither does an unpaired user with no env file. + """ + + def __init__(self, channel=None, serial_resolver=None): + self._channel = channel + self._resolve = serial_resolver or resolve_serial_via_info + self._lock = threading.Lock() + self._resolved = False + self._token: Optional[str] = None + + def _token_for_call(self) -> Optional[str]: + with self._lock: + if self._resolved: + return self._token + self._resolved = True + + if not has_any_tokens(): + # Nothing paired anywhere: do not pay for an Info call. + return None + + serial = self._resolve(self._channel) + if not serial: + logger.debug("No device serial available; sending no token") + return None + + self._token = token_for_serial(serial) + if self._token is None: + logger.debug("No token stored for serial %s", serial) + return self._token + + def _with_token(self, call_details): + if call_details.method in OPEN_METHODS: + return call_details + + token = self._token_for_call() + if not token: + return call_details + + metadata = list(call_details.metadata or []) + metadata.append((METADATA_KEY, f"{METADATA_SCHEME}{token}")) + return _ClientCallDetails(call_details, metadata) + + def intercept_unary_unary(self, continuation, call_details, request): + return continuation(self._with_token(call_details), request) + + def intercept_unary_stream(self, continuation, call_details, request): + return continuation(self._with_token(call_details), request) + + def intercept_stream_unary(self, continuation, call_details, request_iterator): + return continuation(self._with_token(call_details), request_iterator) + + def intercept_stream_stream(self, continuation, call_details, request_iterator): + return continuation(self._with_token(call_details), request_iterator) + + +class _ClientCallDetails(grpc.ClientCallDetails): + """grpc.ClientCallDetails is not always a namedtuple, so carry a copy.""" + + def __init__(self, original, metadata): + self.method = original.method + self.timeout = original.timeout + self.metadata = metadata + self.credentials = original.credentials + self.wait_for_ready = getattr(original, "wait_for_ready", None) + self.compression = getattr(original, "compression", None) + + +def intercept(channel): + """Wrap `channel` so token-gated calls carry the pairing token.""" + return grpc.intercept_channel(channel, AuthInterceptor(channel)) diff --git a/synapse/client/device.py b/synapse/client/device.py index 2350f7b2..13502484 100644 --- a/synapse/client/device.py +++ b/synapse/client/device.py @@ -22,6 +22,7 @@ ListAppsRequest, ListAppsResponse, ) +from synapse.client import auth from synapse.client.config import Config from synapse.utils.log import log_level_to_pb @@ -38,7 +39,10 @@ def __init__(self, uri, verbose=False): self.uri = uri self.channel = grpc.insecure_channel(self.uri) - self.rpc = SynapseDeviceStub(self.channel) + # The interceptor resolves the serial and attaches the pairing token + # lazily, on the first call that needs one. Construction stays free of + # network I/O, so offline construction and existing tests are unaffected. + self.rpc = SynapseDeviceStub(auth.intercept(self.channel)) self.logger = logging.getLogger(__name__) level = logging.DEBUG if verbose else logging.ERROR diff --git a/synapse/client/files.py b/synapse/client/files.py new file mode 100644 index 00000000..7e26b2dc --- /dev/null +++ b/synapse/client/files.py @@ -0,0 +1,176 @@ +"""File transfer over gRPC. + +Replaces SFTP as the transport for device files. The point is not convenience: +SFTP needed its own password, so a device carried two independent credentials. +Going through the SynapseDevice service means file access is governed by the +same pairing token as everything else, and the SFTP password disappears. + +Paths are relative to the device's data root and cannot escape it. That is the +same tree the scifi-sftp account was chrooted to, so reach is unchanged -- what +goes away is the second credential. +""" + +import json +import os +from typing import Callable, Iterator, List, Optional, Tuple + +import grpc + +from synapse.api.files_pb2 import ( + DeleteFileRequest, + ListFilesRequest, + ListFilesResponse, + ReadFileRequest, + WriteFileRequest, +) + +# Chunk size for uploads. Well under gRPC's 4 MB default message limit, leaving +# room for the message's other fields. +CHUNK_BYTES = 1 << 20 # 1 MiB + +ProgressFn = Callable[[int, int], None] + + +def list_files(device, path: str = "", recursive: bool = False) -> List[ListFilesResponse.File]: + return list(device.rpc.ListFiles(ListFilesRequest(path=path, recursive=recursive)).files) + + +PARTIAL_SUFFIX = ".partial" +# Beside each .partial: which version of the device file it holds, as +# {"size": ..., "modified": ...}. Same format nexus-desktop writes. +STAMP_SUFFIX = ".partial.json" + +# (size, modified) of a device file, as listed by ListFiles. +Stamp = Tuple[int, int] + + +def stamp_of(entry: ListFilesResponse.File) -> Stamp: + return (int(entry.size_bytes), int(entry.modified)) + + +def _resume_offset(local_path: str, stamp: Optional[Stamp]) -> int: + """Bytes already downloaded of this exact version of the file, else 0. + + A partial is only worth continuing if the device's file is unchanged since + it was started; resuming onto a replaced file would splice two different + files together. Without a stamp to compare against, the partial's size is + trusted as before. + """ + partial = local_path + PARTIAL_SUFFIX + if not os.path.exists(partial): + return 0 + if stamp is None: + return os.path.getsize(partial) + try: + with open(local_path + STAMP_SUFFIX) as f: + saved = json.load(f) + if (saved["size"], saved["modified"]) == stamp: + return os.path.getsize(partial) + except (OSError, ValueError, KeyError, TypeError): + pass + return 0 # stale or unidentified: start over + + +def _discard_partial(local_path: str) -> None: + for suffix in (PARTIAL_SUFFIX, STAMP_SUFFIX): + try: + os.remove(local_path + suffix) + except FileNotFoundError: + pass + + +def read_file( + device, + remote_path: str, + local_path: str, + progress: Optional[ProgressFn] = None, + resume: bool = True, + stamp: Optional[Stamp] = None, +) -> int: + """Download `remote_path` to `local_path`, returning bytes written. + + Bytes land in `.partial`, which is renamed over `local_path` + only once the download completes. So `local_path` is never half-written, + and resuming is unambiguous: only a leftover `.partial` is continued, never + a complete file that happens to share the name. Pass the device file's + `stamp` (see stamp_of) so a partial of an older version is restarted + rather than continued. Set resume=False to discard any `.partial` and start + over. + """ + partial = local_path + PARTIAL_SUFFIX + start_offset = _resume_offset(local_path, stamp) if resume else 0 + os.makedirs(os.path.dirname(os.path.abspath(local_path)), exist_ok=True) + if stamp is not None: + with open(local_path + STAMP_SUFFIX, "w") as f: + json.dump({"size": stamp[0], "modified": stamp[1]}, f) + elif os.path.exists(local_path + STAMP_SUFFIX): + os.remove(local_path + STAMP_SUFFIX) # it would describe another download + + try: + try: + written = _fetch(device, remote_path, partial, start_offset, progress) + except grpc.RpcError as e: + # The device's copy is now shorter than what we had: it changed since + # the interrupted download, so that partial is from another file. + if start_offset == 0 or e.code() != grpc.StatusCode.OUT_OF_RANGE: + raise + written = _fetch(device, remote_path, partial, 0, progress) + except BaseException: + # The partial is opened before the first chunk (so an empty file still + # gets created), but the device's errors, such as "no such file", only + # arrive once reading starts. Nothing received means nothing to resume. + if not os.path.exists(partial) or os.path.getsize(partial) == 0: + _discard_partial(local_path) + raise + + os.replace(partial, local_path) + _discard_partial(local_path) # just the stamp, now + return written + + +def _fetch(device, remote_path, partial, start_offset, progress) -> int: + stream = device.rpc.ReadFile(ReadFileRequest(path=remote_path, start_offset_bytes=start_offset)) + written = start_offset + # Opened before the first chunk: the server sends zero chunks for an empty + # file, and that file still has to exist afterwards. + with open(partial, "r+b" if start_offset else "wb") as f: + f.seek(start_offset) + f.truncate() + for chunk in stream: + f.write(chunk.data) + written += len(chunk.data) + if progress: + progress(written, chunk.file_size_bytes) + return written + + +def _upload_messages(remote_path: str, local_path: str, progress: Optional[ProgressFn]): + """The request stream: path first, then chunks. + + The first message carries the path and no data, so the server can resolve + and open the target before any bytes arrive. + """ + total = os.path.getsize(local_path) + yield WriteFileRequest(path=remote_path) + sent = 0 + with open(local_path, "rb") as f: + while True: + data = f.read(CHUNK_BYTES) + if not data: + break + sent += len(data) + if progress: + progress(sent, total) + yield WriteFileRequest(data=data) + + +def write_file( + device, local_path: str, remote_path: str, progress: Optional[ProgressFn] = None +) -> int: + """Upload `local_path` to `remote_path`, returning bytes the device wrote.""" + response = device.rpc.WriteFile(_upload_messages(remote_path, local_path, progress)) + return response.bytes_written + + +def delete_file(device, path: str, recursive: bool = False) -> None: + device.rpc.DeleteFile(DeleteFileRequest(path=path, recursive=recursive)) diff --git a/synapse/client/sftp.py b/synapse/client/sftp.py index c4d55253..683e61c1 100644 --- a/synapse/client/sftp.py +++ b/synapse/client/sftp.py @@ -1,7 +1,14 @@ +import getpass import logging +import os import socket +from typing import Optional + import paramiko import paramiko.ssh_exception +import yaml +from rich.console import Console +from rich.prompt import Confirm def connect_sftp(hostname, username, password=None, pass_filename=None, key_filename=None, port=22): """ @@ -64,3 +71,115 @@ def close_sftp(ssh, sftp): sftp.close() if ssh is not None: ssh.close() + +# --- SFTP connection and password handling --------------------------------- +# +# Only for device firmware that predates the gRPC file RPCs. `synapsectl file` +# and `deploy-model` use the file RPCs, and fall back to SFTP (with the +# device's separate file-transfer password) when the device answers them with +# UNIMPLEMENTED -- see _uses_sftp in synapse/cli/files.py. When that firmware +# is no longer in the field, all of this goes, along with the SFTP password. + +SCIFI_DEFAULT_SFTP_USER = "scifi-sftp" +DEFAULT_ENV_FILE = ".scienv" + +def setup_connection( + uri: str, + username: str, + env_file: str, + forget_password: bool, + console: Console, +) -> Optional[tuple[paramiko.SSHClient, paramiko.SFTPClient]]: + from synapse import Device # here, not at the top: synapse/__init__ imports this package + + dev_name = Device(uri).get_name() + password = find_password( + dev_name, env_file + ) # Check if password is provided or stored in env file + if password is None: + console.print(f"[bold red]Didnt find any password for {uri}[/bold red]") + return + + # Open SFTP connection + with console.status("Connecting to Synapse device...", spinner="bouncingBall"): + try: + ssh, sftp_conn = connect_sftp(uri, username, password) + except paramiko.ssh_exception.AuthenticationException: + console.print(f"[bold red]Authentication failed for {uri}[/bold red]") + console.print("[yellow] Incorrect username or password.") + return None + if ssh is None or sftp_conn is None: + console.print(f"[bold red]Failed to connect to {uri}[/bold red]") + return + # If the connection is successful, we can prompt the user if they want to save the password + if not forget_password and dev_name is not None: + save_password(password, env_file, dev_name) + return ssh, sftp_conn + +def find_password(dev_name: Optional[str], env_file: Optional[str]): + password = None + if env_file is not None and dev_name is not None: + if os.path.exists(env_file): + password = load_pass_from_env_file(env_file, dev_name) + if password is not None: + return password + password = getpass.getpass("Enter password: ") + return password + +def save_password(password: str, env_file: str, device_name: str): + if env_file is None or device_name is None or password is None: + return + if os.path.exists(env_file): + if load_pass_from_env_file(env_file, device_name) is not None: + return + + save_pass = Confirm.ask(f"Save password for {device_name} in {env_file}?") + if not save_pass: + return + store_pass_to_env_file(env_file, device_name, password) + +def load_pass_from_env_file(env_file: str, device_name: str) -> Optional[str]: + try: + with open(env_file, "r") as f: + env_loaded = yaml.safe_load(f) + if env_loaded is None: + return None + if "sftp_passwords" not in env_loaded: + return None + passwords = env_loaded.get("sftp_passwords", {}) + if device_name in passwords: + try: + stored_pass = passwords[device_name] + return stored_pass + except Exception: + print( + f"Couldnt read password for {device_name} from env file. Env file may be improperly formatted." + ) + return None + else: + return None + except Exception as e: + print(f"Couldnt read env file at: {env_file}. {e}") + return None + + +# Store password to the .env file in yaml format + +def store_pass_to_env_file(env_file: str, device_name: str, password: str): + try: + prev_env = {} + if os.path.exists(env_file): + with open(env_file, "r") as f: + prev_env = yaml.safe_load(f) + prev_env = prev_env if prev_env is not None else {} + passwords = prev_env.get("sftp_passwords", {}) + passwords[device_name] = password + prev_env["sftp_passwords"] = passwords + with open(env_file, "w", encoding="utf8") as f: + yaml.dump(prev_env, f, default_flow_style=False) + except TypeError: + print( + f"Failed to store pass to env file at: {env_file}. Env file may be improperly formatted." + ) + except Exception as e: + print(f"Failed to store pass to env file at: {env_file}. {e}") diff --git a/synapse/tests/cli/test_auth.py b/synapse/tests/cli/test_auth.py new file mode 100644 index 00000000..4e7ae6f2 --- /dev/null +++ b/synapse/tests/cli/test_auth.py @@ -0,0 +1,324 @@ +"""Tests for `synapsectl pair` / `unpair`. + +The `pair` handshake itself needs a live device on the other end of a gRPC +stream (challenge code, then a human tapping Allow) -- there is no meaningful +way to unit test that without faking most of grpc, so it is left to the +on-device validation task. What's tested here: + + - argparse registration: `pair` and `unpair` (including its optional + positional `identifier`) show up wired to the right functions with the + right flags -- exactly what a --help typo would not be caught by any + other test. + - `unpair`'s two paths, both reachable through the real CLI: + * `unpair ` -- purely local, no network at all. Covers + matching by serial, matching by name, no match, and an ambiguous + name match. + * `unpair --uri ` (no positional) -- contacts the device via + the open `Info` RPC for the authoritative serial, then revokes the + token there with `RevokeAuth` before forgetting it. Covers + found, not found, unreachable, and a revoke that fails. + * both given -- the positional wins and the device is never contacted. + * neither given -- a usage hint instead of a crash. +""" + +import argparse +import contextlib + +import grpc +import pytest + +from synapse.cli import auth as auth_module +from synapse.client import auth as client_auth + + +@pytest.fixture +def env_file(tmp_path, monkeypatch): + path = tmp_path / ".scifi-env" + monkeypatch.setenv("SCIFI_ENV_FILE", str(path)) + return path + + +@pytest.fixture +def console_log(monkeypatch): + """Replace rich's Console with something that just records plain text.""" + lines = [] + + class _FakeConsole: + def print(self, *args, **kwargs): + lines.append(" ".join(str(a) for a in args)) + + def status(self, *args, **kwargs): + return contextlib.nullcontext() + + monkeypatch.setattr(auth_module, "Console", lambda: _FakeConsole()) + return lines + + +class _FakeRpcError(grpc.RpcError): + def __init__(self, code, details): + self._code = code + self._details = details + + def code(self): + return self._code + + def details(self): + return self._details + + +class _FakeInfo: + def __init__(self, serial, name): + self.serial = serial + self.name = name + + +class _FakeRpc: + def __init__(self, info=None, error=None, revoke_error=None): + self._info = info + self._error = error + self._revoke_error = revoke_error + self.revoke_calls = 0 + + def Info(self, request, timeout=None): + if self._error is not None: + raise self._error + return self._info + + def RevokeAuth(self, request, timeout=None): + if self._revoke_error is not None: + raise self._revoke_error + self.revoke_calls += 1 + + +class _FakeDevice: + """Stands in for `syn.Device(uri, verbose)`.""" + + def __init__(self, info=None, error=None, revoke_error=None): + self.rpc = _FakeRpc(info=info, error=error, revoke_error=revoke_error) + + +def _args(uri=None, verbose=False, identifier=None, label=None): + return argparse.Namespace( + uri=uri, verbose=verbose, identifier=identifier, label=label + ) + + +def _patch_device(monkeypatch, device): + monkeypatch.setattr(auth_module.syn, "Device", lambda uri, verbose: device) + + +def _forbid_device_contact(monkeypatch): + """Fail the test if `unpair` tries to construct a Device at all.""" + + def _fail(uri, verbose): + raise AssertionError("unpair should not have contacted the device") + + monkeypatch.setattr(auth_module.syn, "Device", _fail) + + +# --- argparse registration ------------------------------------------------- + + +def test_pair_and_unpair_are_registered(): + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers(title="Commands") + auth_module.add_commands(subparsers) + + pair_args = parser.parse_args(["pair"]) + assert pair_args.func is auth_module.pair + assert pair_args.label is None + + pair_args = parser.parse_args(["pair", "--label", "calvin@laptop"]) + assert pair_args.label == "calvin@laptop" + + unpair_args = parser.parse_args(["unpair"]) + assert unpair_args.func is auth_module.unpair + assert unpair_args.identifier is None + + unpair_args = parser.parse_args(["unpair", "sci-fi-1234"]) + assert unpair_args.func is auth_module.unpair + assert unpair_args.identifier == "sci-fi-1234" + + +# --- unpair : local-only, no network ---------------------------- + + +def test_unpair_by_identifier_matches_serial(env_file, console_log, monkeypatch): + client_auth.save_token("NYX1512-0042", "sci-fi-1234", "f3a9c1") + _forbid_device_contact(monkeypatch) + + auth_module.unpair(_args(identifier="NYX1512-0042")) + + assert client_auth.token_for_serial("NYX1512-0042") is None + assert any("Forgot the token" in line for line in console_log) + + +def test_unpair_by_identifier_matches_name(env_file, console_log, monkeypatch): + client_auth.save_token("NYX1512-0042", "sci-fi-1234", "f3a9c1") + _forbid_device_contact(monkeypatch) + + auth_module.unpair(_args(identifier="sci-fi-1234")) + + assert client_auth.token_for_serial("NYX1512-0042") is None + assert any("Forgot the token" in line for line in console_log) + + +def test_unpair_by_identifier_no_match_lists_stored_entries( + env_file, console_log, monkeypatch +): + client_auth.save_token("NYX1512-0042", "sci-fi-1234", "f3a9c1") + _forbid_device_contact(monkeypatch) + + auth_module.unpair(_args(identifier="nonexistent")) + + # Nothing removed, and the real entry is surfaced so the user can see the + # right spelling. + assert client_auth.token_for_serial("NYX1512-0042") == "f3a9c1" + assert any("No stored entry matches" in line for line in console_log) + assert any("sci-fi-1234" in line for line in console_log) + + +def test_unpair_by_identifier_ambiguous_name_removes_nothing( + env_file, console_log, monkeypatch +): + client_auth.save_token("NYX1512-0042", "dup-name", "token-a") + client_auth.save_token("NYX1512-0099", "dup-name", "token-b") + _forbid_device_contact(monkeypatch) + + auth_module.unpair(_args(identifier="dup-name")) + + # Nothing removed: we refuse to guess between the two matches. + assert client_auth.token_for_serial("NYX1512-0042") == "token-a" + assert client_auth.token_for_serial("NYX1512-0099") == "token-b" + assert any("matches more than one" in line for line in console_log) + assert any("NYX1512-0042" in line for line in console_log) + assert any("NYX1512-0099" in line for line in console_log) + + +# --- unpair --uri : contacts the device for the authoritative serial + + +def test_unpair_via_uri_revokes_on_the_device_then_forgets( + env_file, console_log, monkeypatch +): + client_auth.save_token("NYX1512-0042", "sci-fi-1234", "f3a9c1") + device = _FakeDevice(info=_FakeInfo(serial="NYX1512-0042", name="sci-fi-1234")) + _patch_device(monkeypatch, device) + + auth_module.unpair(_args(uri="10.0.0.5")) + + assert device.rpc.revoke_calls == 1 + assert client_auth.token_for_serial("NYX1512-0042") is None + assert any("Revoked" in line for line in console_log) + + +@pytest.mark.parametrize( + "code", [grpc.StatusCode.UNAUTHENTICATED, grpc.StatusCode.UNIMPLEMENTED] +) +def test_unpair_via_uri_forgets_when_the_device_has_nothing_to_revoke( + env_file, console_log, monkeypatch, code +): + client_auth.save_token("NYX1512-0042", "sci-fi-1234", "f3a9c1") + _patch_device( + monkeypatch, + _FakeDevice( + info=_FakeInfo(serial="NYX1512-0042", name="sci-fi-1234"), + revoke_error=_FakeRpcError(code, "x"), + ), + ) + + auth_module.unpair(_args(uri="10.0.0.5")) + + assert client_auth.token_for_serial("NYX1512-0042") is None + + +def test_unpair_via_uri_keeps_the_token_when_revoke_fails(env_file, console_log, monkeypatch): + client_auth.save_token("NYX1512-0042", "sci-fi-1234", "f3a9c1") + _patch_device( + monkeypatch, + _FakeDevice( + info=_FakeInfo(serial="NYX1512-0042", name="sci-fi-1234"), + revoke_error=_FakeRpcError(grpc.StatusCode.UNAVAILABLE, "dropped"), + ), + ) + + auth_module.unpair(_args(uri="10.0.0.5")) + + assert client_auth.token_for_serial("NYX1512-0042") == "f3a9c1" + assert any("Could not revoke" in line for line in console_log) + + +def test_unpair_via_uri_reports_no_token_when_device_never_paired( + env_file, console_log, monkeypatch +): + _patch_device( + monkeypatch, _FakeDevice(info=_FakeInfo(serial="NYX1512-0042", name="sci-fi-1234")) + ) + + auth_module.unpair(_args(uri="10.0.0.5")) + + assert client_auth.load_tokens() == {} + assert any("No stored token" in line for line in console_log) + + +def test_unpair_via_uri_when_device_unreachable_does_not_guess( + env_file, console_log, monkeypatch +): + client_auth.save_token("NYX1512-0042", "sci-fi-1234", "f3a9c1") + _patch_device( + monkeypatch, + _FakeDevice(error=_FakeRpcError(grpc.StatusCode.UNAVAILABLE, "no route")), + ) + + auth_module.unpair(_args(uri="10.0.0.5")) + + # Nothing removed -- the old silent name-fallback against --uri is gone; + # the message instead points at `unpair `. + assert client_auth.token_for_serial("NYX1512-0042") == "f3a9c1" + assert any("Could not reach the device" in line for line in console_log) + assert any("unpair " in line for line in console_log) + + +# --- both / neither given --------------------------------------------------- + + +def test_unpair_prefers_identifier_and_never_contacts_device_when_both_given( + env_file, console_log, monkeypatch +): + client_auth.save_token("NYX1512-0042", "sci-fi-1234", "f3a9c1") + _forbid_device_contact(monkeypatch) + + auth_module.unpair(_args(uri="10.0.0.5", identifier="sci-fi-1234")) + + assert client_auth.token_for_serial("NYX1512-0042") is None + assert any("Forgot the token" in line for line in console_log) + + +def test_unpair_with_neither_uri_nor_identifier_prints_hint( + env_file, console_log, monkeypatch +): + _forbid_device_contact(monkeypatch) + + auth_module.unpair(_args()) + + assert any("Specify a device" in line for line in console_log) + +# --- pair: interrupt before any code is on screen --------------------------- + + +def test_pair_ctrl_c_during_info_is_not_a_traceback( + monkeypatch, console_log, env_file +): + """Ctrl-C in the 10s Info window must read as a cancellation, not a crash. + + The interrupt during the *approval wait* was always handled. This is the + earlier window -- before a code exists -- which used to escape as a raw + KeyboardInterrupt traceback. Nothing has been asked of the device yet, so + there is nothing to withdraw and nothing to save. + """ + _patch_device(monkeypatch, _FakeDevice(error=KeyboardInterrupt())) + + auth_module.pair(_args(uri="10.0.0.1")) + + assert any("cancelled" in line.lower() for line in console_log), console_log + assert not env_file.exists(), "an interrupted pair must not write a token" diff --git a/synapse/tests/cli/test_files.py b/synapse/tests/cli/test_files.py new file mode 100644 index 00000000..3900e244 --- /dev/null +++ b/synapse/tests/cli/test_files.py @@ -0,0 +1,295 @@ +"""`synapsectl file get` must ask before replacing a local file.""" + +import argparse + +from synapse.api.files_pb2 import ReadFileResponse +from synapse.cli import files as files_cli + + +class _FakeRpc: + def __init__(self): + self.read_requests = [] + + def ListFiles(self, request): + from synapse.api.files_pb2 import ListFilesResponse + + return ListFilesResponse(files=[ListFilesResponse.File(path="a.h5", size_bytes=6)]) + + def ReadFile(self, request): + self.read_requests.append(request) + return iter([ReadFileResponse(path="a.h5", data=b"remote", file_size_bytes=6)]) + + +class _FakeDevice: + def __init__(self, *args, **kwargs): + self.rpc = _FakeRpc() + + +def _get(monkeypatch, tmp_path, answer, yes=False): + device = _FakeDevice() + monkeypatch.setattr(files_cli, "Device", lambda *a, **k: device) + monkeypatch.setattr(files_cli.Confirm, "ask", lambda *a, **k: answer) + args = argparse.Namespace( + uri="x", verbose=False, remote_path="a.h5", output_path=str(tmp_path), + recursive=False, no_resume=False, yes=yes, + username=None, env_file=None, forget_password=False, + ) + files_cli.get(args) + return device + + +def test_declining_the_prompt_leaves_the_local_file_and_fetches_nothing(monkeypatch, tmp_path): + (tmp_path / "a.h5").write_bytes(b"mine") + device = _get(monkeypatch, tmp_path, answer=False) + assert device.rpc.read_requests == [] + assert (tmp_path / "a.h5").read_bytes() == b"mine" + + +def test_accepting_the_prompt_replaces_the_local_file(monkeypatch, tmp_path): + (tmp_path / "a.h5").write_bytes(b"mine") + _get(monkeypatch, tmp_path, answer=True) + assert (tmp_path / "a.h5").read_bytes() == b"remote" + + +def test_yes_overwrites_without_asking(monkeypatch, tmp_path): + (tmp_path / "a.h5").write_bytes(b"mine") + _get(monkeypatch, tmp_path, answer=False, yes=True) # would decline, if it were asked + assert (tmp_path / "a.h5").read_bytes() == b"remote" + + +class _FakePutRpc: + """A device holding `a.h5` at its data root.""" + + def __init__(self): + self.uploads = [] + + def ListFiles(self, request): + from synapse.api.files_pb2 import ListFilesResponse + + return ListFilesResponse(files=[ListFilesResponse.File(path="a.h5", size_bytes=4)]) + + def WriteFile(self, messages): + from synapse.api.files_pb2 import WriteFileResponse + + msgs = list(messages) + self.uploads.append(msgs[0].path) + return WriteFileResponse(path=msgs[0].path, bytes_written=sum(len(m.data) for m in msgs)) + + +def _put(monkeypatch, tmp_path, answer, yes=False, remote=None): + device = _FakeDevice() + device.rpc = _FakePutRpc() + monkeypatch.setattr(files_cli, "Device", lambda *a, **k: device) + monkeypatch.setattr(files_cli.Confirm, "ask", lambda *a, **k: answer) + src = tmp_path / "a.h5" + src.write_bytes(b"mine") + args = argparse.Namespace( + uri="x", verbose=False, local_path=str(src), remote_path=remote, recursive=False, yes=yes + ) + files_cli.put(args) + return device.rpc.uploads + + +def test_put_declined_does_not_upload_over_a_device_file(monkeypatch, tmp_path): + assert _put(monkeypatch, tmp_path, answer=False) == [] + + +def test_put_accepted_or_yes_uploads_over_a_device_file(monkeypatch, tmp_path): + assert _put(monkeypatch, tmp_path, answer=True) == ["a.h5"] + assert _put(monkeypatch, tmp_path, answer=False, yes=True) == ["a.h5"] + + +def test_put_to_a_new_name_does_not_ask(monkeypatch, tmp_path): + assert _put(monkeypatch, tmp_path, answer=False, remote="b.h5") == ["b.h5"] + + +def test_an_interrupted_get_says_where_the_partial_is(monkeypatch, tmp_path, capsys): + import pytest + + class _InterruptingRpc(_FakeRpc): + def ReadFile(self, request): + yield ReadFileResponse(path="a.h5", data=b"half", file_size_bytes=8) + raise KeyboardInterrupt + + device = _FakeDevice() + device.rpc = _InterruptingRpc() + monkeypatch.setattr(files_cli, "Device", lambda *a, **k: device) + args = argparse.Namespace( + uri="x", verbose=False, remote_path="a.h5", output_path=str(tmp_path), + recursive=False, no_resume=False, yes=False, + username=None, env_file=None, forget_password=False, + ) + with pytest.raises(SystemExit): + files_cli.get(args) + + out = capsys.readouterr().out + assert "a.h5.partial" in out and "resume" in out + assert (tmp_path / "a.h5.partial").read_bytes() == b"half" + + +def test_ls_hides_dotfiles_unless_all(monkeypatch): + from synapse.api.files_pb2 import ListFilesResponse + + printed = [] + monkeypatch.setattr(files_cli, "_print_file_list", lambda entries, console: printed.append([f.path for f in entries])) + + class _Rpc: + def ListFiles(self, request): + F = ListFilesResponse.File + return ListFilesResponse(files=[ + F(path="rec/a.h5"), F(path="rec/.scifi-checked"), F(path="rec/.cache", is_dir=True), + F(path="rec/.cache/x"), + ]) + + device = _FakeDevice() + device.rpc = _Rpc() + monkeypatch.setattr(files_cli, "Device", lambda *a, **k: device) + base = dict(uri="x", verbose=False, path="rec", recursive=True, username=None, env_file=None, forget_password=False) + + files_cli.ls(argparse.Namespace(**base, all=False)) + files_cli.ls(argparse.Namespace(**base, all=True)) + + assert printed[0] == ["rec/a.h5"] + assert len(printed[1]) == 4 + + +def test_listing_a_hidden_directory_by_name_shows_its_contents(): + assert not files_cli._is_hidden(".cache/x", ".cache") + assert files_cli._is_hidden(".cache/x", "") + + +def _get_missing(monkeypatch, tmp_path, listing): + import pytest + from synapse.api.files_pb2 import ListFilesResponse + + class _Rpc(_FakeRpc): + def ListFiles(self, request): + return ListFilesResponse(files=listing) + + device = _FakeDevice() + device.rpc = _Rpc() + monkeypatch.setattr(files_cli, "Device", lambda *a, **k: device) + args = argparse.Namespace( + uri="x", verbose=False, remote_path="a.h5", output_path=str(tmp_path), + recursive=False, no_resume=False, yes=False, + username=None, env_file=None, forget_password=False, + ) + with pytest.raises(SystemExit) as exit_info: + files_cli.get(args) + return device, exit_info.value.code + + +def test_get_of_a_missing_file_says_so_and_touches_nothing(monkeypatch, tmp_path, capsys): + device, code = _get_missing(monkeypatch, tmp_path, listing=[]) + assert code == 1 + assert "a.h5 does not exist on the device" in capsys.readouterr().out + assert device.rpc.read_requests == [] + assert list(tmp_path.iterdir()) == [] + + +def test_get_of_a_directory_points_at_recursive(monkeypatch, tmp_path, capsys): + from synapse.api.files_pb2 import ListFilesResponse + + _, code = _get_missing(monkeypatch, tmp_path, listing=[ListFilesResponse.File(path="a.h5", is_dir=True)]) + assert code == 1 + assert "--recursive" in capsys.readouterr().out + + +class _OldFirmwareRpc: + """Device firmware from before the file RPCs existed.""" + + def ListFiles(self, request): + class _Unimplemented(grpc.RpcError): + def code(self): + return grpc.StatusCode.UNIMPLEMENTED + + raise _Unimplemented() + + +import grpc # noqa: E402 (used by the fake above) +import pytest # noqa: E402 + + +@pytest.mark.parametrize("command", ["ls", "get", "put", "rm"]) +def test_old_firmware_falls_back_to_sftp_and_says_to_update(monkeypatch, capsys, command): + device = _FakeDevice() + device.rpc = _OldFirmwareRpc() + monkeypatch.setattr(files_cli, "Device", lambda *a, **k: device) + called = [] + monkeypatch.setattr(files_cli.files_sftp, command, lambda args: called.append(args)) + args = argparse.Namespace(uri="x", verbose=False, recursive=False, yes=False) + + getattr(files_cli, command)(args) + + assert called == [args], f"{command} did not hand over to the SFTP implementation" + out = capsys.readouterr().out + assert "predates device pairing" in out and "Update the device" in out + + +def test_new_firmware_does_not_use_sftp(monkeypatch): + device = _FakeDevice() # its ListFiles answers normally + monkeypatch.setattr(files_cli, "Device", lambda *a, **k: device) + monkeypatch.setattr(files_cli.files_sftp, "ls", lambda args: pytest.fail("used SFTP on new firmware")) + files_cli.ls(argparse.Namespace(uri="x", verbose=False, path="", recursive=False, all=False)) + + +def test_sftp_ls_with_no_path_lists_the_root(monkeypatch): + listed = [] + + class _Sftp: + def listdir_attr(self, path): + listed.append(path) + return [] + + monkeypatch.setattr(files_cli.files_sftp.sftp, "setup_connection", lambda *a: (None, _Sftp())) + monkeypatch.setattr(files_cli.files_sftp.sftp, "close_sftp", lambda *a: None) + files_cli.files_sftp.ls( + argparse.Namespace(uri="x", username="u", env_file=".scienv", forget_password=False, path="") + ) + assert listed == ["/"] + + +class _ExistingSftp: + """Every remote path exists as a 1-byte regular file.""" + + def __init__(self): + self.transfers = [] + + def stat(self, path): + import paramiko + + a = paramiko.SFTPAttributes() + a.st_mode, a.st_size = 0o100644, 1 + return a + + def get(self, remote, local, callback=None): + self.transfers.append(("get", remote)) + + def put(self, local, remote, callback=None): + self.transfers.append(("put", remote)) + + +@pytest.mark.parametrize("answer, overwrite, expect", [(False, False, 0), (True, False, 1), (False, True, 1)]) +def test_sftp_get_and_put_ask_before_replacing(monkeypatch, tmp_path, answer, overwrite, expect): + from rich.console import Console + + from synapse.cli import files_sftp + + monkeypatch.setattr(files_cli.Confirm, "ask", lambda *a, **k: answer) + local = tmp_path / "a.h5" + local.write_bytes(b"x") + conn = _ExistingSftp() + + files_sftp.get_file(conn, "disk_writer/a.h5", str(local), Console(), overwrite) + files_sftp.put_file(conn, str(local), "disk_writer/a.h5", Console(), overwrite) + + assert len(conn.transfers) == 2 * expect + + +def test_old_firmware_recursive_rm_asks_first(monkeypatch): + device = _FakeDevice() + device.rpc = _OldFirmwareRpc() + monkeypatch.setattr(files_cli, "Device", lambda *a, **k: device) + monkeypatch.setattr(files_cli.Confirm, "ask", lambda *a, **k: False) + monkeypatch.setattr(files_cli.files_sftp, "rm", lambda args: pytest.fail("deleted without asking")) + files_cli.rm(argparse.Namespace(uri="x", verbose=False, path="disk_writer", recursive=True, yes=False)) diff --git a/synapse/tests/client/test_auth.py b/synapse/tests/client/test_auth.py new file mode 100644 index 00000000..75b929cd --- /dev/null +++ b/synapse/tests/client/test_auth.py @@ -0,0 +1,246 @@ +import logging +import os +import stat + +import pytest + +from synapse.client import auth + +logger = logging.getLogger(__name__) + + +@pytest.fixture +def env_file(tmp_path, monkeypatch): + path = tmp_path / ".scifi-env" + monkeypatch.setenv("SCIFI_ENV_FILE", str(path)) + return path + + +def test_missing_file_yields_no_tokens(env_file): + assert auth.load_tokens() == {} + assert auth.token_for_serial("NYX1512-0042") is None + assert auth.has_any_tokens() is False + + +def test_saved_token_round_trips(env_file): + auth.save_token("NYX1512-0042", "sci-fi-1234", "f3a9c1") + + assert auth.token_for_serial("NYX1512-0042") == "f3a9c1" + assert auth.has_any_tokens() is True + assert auth.load_tokens() == {"NYX1512-0042": ("sci-fi-1234", "f3a9c1")} + + +def test_renamed_device_still_matches_by_serial(env_file): + auth.save_token("NYX1512-0042", "old-name", "f3a9c1") + + # The device was renamed on a later pairing; the serial (and token) are + # unchanged, so the lookup must still work, and the stored hint must be + # the new name, not the stale one. + auth.save_token("NYX1512-0042", "new-name", "f3a9c1") + + assert auth.token_for_serial("NYX1512-0042") == "f3a9c1" + assert auth.load_tokens()["NYX1512-0042"] == ("new-name", "f3a9c1") + + +def test_saving_the_same_serial_replaces_the_entry(env_file): + auth.save_token("NYX1512-0042", "sci-fi-1234", "old-token") + auth.save_token("NYX1512-0042", "renamed", "new-token") + + tokens = auth.load_tokens() + assert len(tokens) == 1 + assert tokens["NYX1512-0042"] == ("renamed", "new-token") + + +def test_multiple_devices_coexist(env_file): + auth.save_token("NYX1512-0042", "sci-fi-1234", "token-a") + auth.save_token("NYX1512-0099", "sci-fi-9999", "token-b") + + assert auth.token_for_serial("NYX1512-0042") == "token-a" + assert auth.token_for_serial("NYX1512-0099") == "token-b" + + +def test_file_is_owner_only(env_file): + auth.save_token("NYX1512-0042", "sci-fi-1234", "f3a9c1") + + mode = stat.S_IMODE(os.stat(env_file).st_mode) + assert mode == 0o600 + + +def test_no_temp_file_left_behind_after_save(env_file): + auth.save_token("NYX1512-0042", "sci-fi-1234", "f3a9c1") + + leftovers = [p.name for p in env_file.parent.iterdir() if p != env_file] + assert leftovers == [] + + +def test_failed_write_leaves_original_file_intact(env_file): + if os.geteuid() == 0: + pytest.skip("root bypasses directory permission checks") + + auth.save_token("NYX1512-0042", "sci-fi-1234", "f3a9c1") + original = env_file.read_bytes() + + # Strip write permission on the directory so the temp file used for the + # atomic replace cannot even be created; the original must survive. + env_file.parent.chmod(0o500) + try: + with pytest.raises(OSError): + auth.save_token("NYX1512-0099", "sci-fi-9999", "token-b") + finally: + env_file.parent.chmod(0o700) + + assert env_file.read_bytes() == original + + +def test_comments_and_blank_lines_are_ignored(env_file): + env_file.write_text( + "# serial name token\n" + "\n" + "NYX1512-0042 sci-fi-1234 f3a9c1\n" + ) + + assert auth.token_for_serial("NYX1512-0042") == "f3a9c1" + + +def test_malformed_lines_are_skipped_not_fatal(env_file): + env_file.write_text( + "this-line-has-only-one-field\n" + "NYX1512-0042 sci-fi-1234 f3a9c1\n" + ) + + assert auth.token_for_serial("NYX1512-0042") == "f3a9c1" + + +def test_remove_token(env_file): + auth.save_token("NYX1512-0042", "sci-fi-1234", "f3a9c1") + + assert auth.remove_token("NYX1512-0042") is True + assert auth.token_for_serial("NYX1512-0042") is None + assert auth.remove_token("NYX1512-0042") is False + + +def test_env_override_is_honored(tmp_path, monkeypatch): + custom = tmp_path / "custom-env" + monkeypatch.setenv("SCIFI_ENV_FILE", str(custom)) + + auth.save_token("NYX1512-0042", "sci-fi-1234", "f3a9c1") + + assert custom.exists() + + +def test_open_methods_are_not_gated(): + for name in ("Info", "GetLogs", "TailLogs", "RequestAuth"): + assert f"/synapse.SynapseDevice/{name}" in auth.OPEN_METHODS + + +def test_closed_methods_are_not_in_the_open_set(): + for name in ("Configure", "Start", "Stop", "Query", "StreamQuery"): + assert f"/synapse.SynapseDevice/{name}" not in auth.OPEN_METHODS + + +class _FakeCallDetails: + def __init__(self, method): + self.method = method + self.timeout = None + self.metadata = None + self.credentials = None + self.wait_for_ready = None + self.compression = None + + +def test_interceptor_sends_no_metadata_for_open_methods(env_file, monkeypatch): + auth.save_token("NYX1512-0042", "sci-fi-1234", "f3a9c1") + + # Resolution would need a network call; if it happens for an open method, + # this raises and the test fails. + def explode(_channel): + raise AssertionError("resolved the serial for an open method") + + interceptor = auth.AuthInterceptor(serial_resolver=explode) + seen = {} + + def continuation(details, request): + seen["metadata"] = details.metadata + return "response" + + result = interceptor.intercept_unary_unary( + continuation, _FakeCallDetails("/synapse.SynapseDevice/Info"), object() + ) + + assert result == "response" + assert not seen["metadata"] + + +def test_interceptor_attaches_token_for_closed_methods(env_file): + auth.save_token("NYX1512-0042", "sci-fi-1234", "f3a9c1") + + interceptor = auth.AuthInterceptor(serial_resolver=lambda _c: "NYX1512-0042") + seen = {} + + def continuation(details, request): + seen["metadata"] = dict(details.metadata or []) + return "response" + + interceptor.intercept_unary_unary( + continuation, _FakeCallDetails("/synapse.SynapseDevice/Configure"), object() + ) + + assert seen["metadata"]["authorization"] == "Bearer f3a9c1" + + +def test_interceptor_skips_resolution_when_no_tokens_exist(env_file): + def explode(_channel): + raise AssertionError("resolved the serial with an empty env file") + + interceptor = auth.AuthInterceptor(serial_resolver=explode) + seen = {} + + def continuation(details, request): + seen["metadata"] = details.metadata + return "response" + + interceptor.intercept_unary_unary( + continuation, _FakeCallDetails("/synapse.SynapseDevice/Configure"), object() + ) + + assert not seen["metadata"] + + +def test_interceptor_resolves_only_once_per_channel(env_file): + auth.save_token("NYX1512-0042", "sci-fi-1234", "f3a9c1") + + calls = [] + + def counting_resolver(_channel): + calls.append(1) + return "NYX1512-0042" + + interceptor = auth.AuthInterceptor(serial_resolver=counting_resolver) + + def continuation(details, request): + return "response" + + for _ in range(3): + interceptor.intercept_unary_unary( + continuation, _FakeCallDetails("/synapse.SynapseDevice/Configure"), object() + ) + + assert len(calls) == 1 + + +def test_unknown_serial_sends_no_token(env_file): + auth.save_token("NYX1512-0042", "sci-fi-1234", "f3a9c1") + + interceptor = auth.AuthInterceptor(serial_resolver=lambda _c: "SOME-OTHER-SERIAL") + seen = {} + + def continuation(details, request): + seen["metadata"] = details.metadata + return "response" + + interceptor.intercept_unary_unary( + continuation, _FakeCallDetails("/synapse.SynapseDevice/Configure"), object() + ) + + # No token for that serial: send nothing and let the server explain. + assert not seen["metadata"] diff --git a/synapse/tests/client/test_files.py b/synapse/tests/client/test_files.py new file mode 100644 index 00000000..f0aec1dc --- /dev/null +++ b/synapse/tests/client/test_files.py @@ -0,0 +1,316 @@ +"""Tests for the gRPC file client. + +The device is faked: these pin the client's side of the contract -- chunking, +the path-first upload framing, and resume via a .partial file -- which is where the bugs that only +show up on multi-gigabyte files would live. +""" + +import os + +import grpc +import pytest + +from synapse.api.files_pb2 import ListFilesResponse, ReadFileResponse, WriteFileResponse +from synapse.client import files as files_client + + +class _FakeRpc: + def __init__(self, read_chunks=None, listing=None): + self._read_chunks = read_chunks or [] + self._listing = listing or [] + self.written = bytearray() + self.upload_messages = [] + self.read_requests = [] + + def ListFiles(self, request): + self.list_request = request + return ListFilesResponse(files=self._listing) + + def ReadFile(self, request): + self.read_requests.append(request) + return iter(self._read_chunks) + + def WriteFile(self, message_iter): + for m in message_iter: + self.upload_messages.append(m) + self.written.extend(m.data) + return WriteFileResponse(path="x", bytes_written=len(self.written)) + + def DeleteFile(self, request): + self.delete_request = request + return None + + +class _FakeDevice: + def __init__(self, **kw): + self.rpc = _FakeRpc(**kw) + + +def test_upload_sends_the_path_first_and_then_chunks(tmp_path): + # The server resolves and opens the target from the first message, so a + # first message carrying data-but-no-path would have nowhere to write. + src = tmp_path / "big.bin" + src.write_bytes(b"a" * (files_client.CHUNK_BYTES + 17)) + + device = _FakeDevice() + written = files_client.write_file(device, str(src), "run_1/big.bin") + + msgs = device.rpc.upload_messages + assert msgs[0].path == "run_1/big.bin" + assert msgs[0].data == b"" + assert all(m.path == "" for m in msgs[1:]), "path must not repeat on every chunk" + assert len(msgs) == 3, "expected path + 2 chunks" + assert bytes(device.rpc.written) == src.read_bytes() + assert written == src.stat().st_size + + +def test_upload_chunks_stay_under_the_grpc_message_limit(tmp_path): + # The whole reason WriteFile is client-streaming: a 4 MB default message + # limit makes a one-shot upload impossible for real recordings. + src = tmp_path / "x.bin" + src.write_bytes(b"b" * (5 * 1024 * 1024)) + device = _FakeDevice() + files_client.write_file(device, str(src), "x.bin") + assert all(len(m.data) <= files_client.CHUNK_BYTES for m in device.rpc.upload_messages) + + +def test_download_writes_every_chunk_in_order(tmp_path): + chunks = [ + ReadFileResponse(path="f", data=b"hello ", start_offset_bytes=0, file_size_bytes=11), + ReadFileResponse(path="f", data=b"world", start_offset_bytes=6, file_size_bytes=11), + ] + device = _FakeDevice(read_chunks=chunks) + dest = tmp_path / "out" / "f" + n = files_client.read_file(device, "f", str(dest), resume=False) + assert dest.read_bytes() == b"hello world" + assert n == 11 + + +def test_download_resumes_an_interrupted_partial(tmp_path): + # A 3 GB recording that dies at 90% must not start over. + dest = tmp_path / "f" + (tmp_path / "f.partial").write_bytes(b"hello ") + chunks = [ReadFileResponse(path="f", data=b"world", start_offset_bytes=6, file_size_bytes=11)] + device = _FakeDevice(read_chunks=chunks) + + n = files_client.read_file(device, "f", str(dest)) + + assert device.rpc.read_requests[0].start_offset_bytes == 6, "did not ask to resume" + assert dest.read_bytes() == b"hello world" + assert not (tmp_path / "f.partial").exists() + assert n == 11 + + +def test_a_complete_local_file_is_replaced_not_appended_to(tmp_path): + # The bug this pins: a same-named local file used to be treated as a + # partial download, so the remote file's tail was appended to it. + dest = tmp_path / "f" + dest.write_bytes(b"other") + chunks = [ReadFileResponse(path="f", data=b"remote!", start_offset_bytes=0, file_size_bytes=7)] + device = _FakeDevice(read_chunks=chunks) + + files_client.read_file(device, "f", str(dest)) + + assert device.rpc.read_requests[0].start_offset_bytes == 0 + assert dest.read_bytes() == b"remote!" + + +def test_an_interrupted_download_leaves_the_destination_untouched(tmp_path): + dest = tmp_path / "f" + dest.write_bytes(b"previous copy") + + def broken(): + yield ReadFileResponse(path="f", data=b"half", start_offset_bytes=0, file_size_bytes=8) + raise ConnectionError("link dropped") + + device = _FakeDevice() + device.rpc.ReadFile = lambda request: broken() + with pytest.raises(ConnectionError): + files_client.read_file(device, "f", str(dest)) + + assert dest.read_bytes() == b"previous copy" + assert (tmp_path / "f.partial").read_bytes() == b"half" + + +def test_a_partial_longer_than_the_remote_file_restarts(tmp_path): + class _OutOfRange(grpc.RpcError): + def code(self): + return grpc.StatusCode.OUT_OF_RANGE + + dest = tmp_path / "f" + (tmp_path / "f.partial").write_bytes(b"from a bigger, older file") + calls = [] + + def read(request): + calls.append(request.start_offset_bytes) + if request.start_offset_bytes: + raise _OutOfRange() + return iter([ReadFileResponse(path="f", data=b"new", start_offset_bytes=0, file_size_bytes=3)]) + + device = _FakeDevice() + device.rpc.ReadFile = read + files_client.read_file(device, "f", str(dest)) + + assert calls == [25, 0] + assert dest.read_bytes() == b"new" + + +def test_resume_false_refetches_from_the_start(tmp_path): + dest = tmp_path / "f" + (tmp_path / "f.partial").write_bytes(b"stale") + chunks = [ReadFileResponse(path="f", data=b"fresh", start_offset_bytes=0, file_size_bytes=5)] + device = _FakeDevice(read_chunks=chunks) + files_client.read_file(device, "f", str(dest), resume=False) + assert device.rpc.read_requests[0].start_offset_bytes == 0 + assert dest.read_bytes() == b"fresh" + + +def test_empty_file_still_creates_the_local_file(tmp_path): + device = _FakeDevice(read_chunks=[]) + dest = tmp_path / "empty" + assert files_client.read_file(device, "empty", str(dest), resume=False) == 0 + assert dest.exists() + + +def test_delete_passes_recursive_through(tmp_path): + device = _FakeDevice() + files_client.delete_file(device, "run_1", recursive=True) + assert device.rpc.delete_request.recursive is True + assert device.rpc.delete_request.path == "run_1" + + +def test_upload_into_a_directory_appends_the_basename(tmp_path, monkeypatch): + """`put file some/dir` must mean into that directory. + + The bug: uploading to an existing directory transferred the whole file and + then failed at the final rename, because the server cannot replace a + directory with a file. 194 MB were sent before the error appeared. + """ + from synapse.api.files_pb2 import ListFilesResponse + from synapse.cli import files as cli + + src = tmp_path / "recording.h5" + src.write_bytes(b"data") + + listing = [ListFilesResponse.File(path="hdf5-replay", size_bytes=0, is_dir=True)] + device = _FakeDevice(listing=listing) + + assert cli._remote_entry(device, "hdf5-replay").is_dir + + captured = {} + + def fake_upload(dev, console, local, remote): + captured["remote"] = remote + return 4 + + monkeypatch.setattr(cli, "_upload_one", fake_upload) + + class Args: + uri = "1.2.3.4" + verbose = False + local_path = str(src) + remote_path = "hdf5-replay" + recursive = False + yes = False + + monkeypatch.setattr(cli, "Device", lambda uri, verbose: device) + cli.put(Args()) + + assert captured["remote"] == "hdf5-replay/recording.h5" + + +def test_upload_to_a_plain_path_is_left_alone(tmp_path, monkeypatch): + from synapse.cli import files as cli + + src = tmp_path / "x.bin" + src.write_bytes(b"y") + device = _FakeDevice(listing=[]) + + captured = {} + monkeypatch.setattr(cli, "_upload_one", lambda d, c, l, r: captured.setdefault("remote", r)) + monkeypatch.setattr(cli, "Device", lambda uri, verbose: device) + + class Args: + uri = "1.2.3.4" + verbose = False + local_path = str(src) + remote_path = "dump/explicit.bin" + recursive = False + + cli.put(Args()) + assert captured["remote"] == "dump/explicit.bin" + + +def test_a_missing_remote_file_leaves_no_empty_partial(tmp_path): + class _NotFound(grpc.RpcError): + def code(self): + return grpc.StatusCode.NOT_FOUND + + def read(request): + raise _NotFound() + yield # a generator, like the real stream: the error surfaces on iteration + + device = _FakeDevice() + device.rpc.ReadFile = read + with pytest.raises(grpc.RpcError): + files_client.read_file(device, "missing.h5", str(tmp_path / "missing.h5")) + + assert list(tmp_path.iterdir()) == [] + + +def _stamped_partial(tmp_path, data, size, modified): + import json + + (tmp_path / "f.partial").write_bytes(data) + (tmp_path / "f.partial.json").write_text(json.dumps({"size": size, "modified": modified})) + + +def test_resumes_when_the_device_file_is_unchanged(tmp_path): + _stamped_partial(tmp_path, b"hello ", size=11, modified=100) + chunks = [ReadFileResponse(path="f", data=b"world", start_offset_bytes=6, file_size_bytes=11)] + device = _FakeDevice(read_chunks=chunks) + + files_client.read_file(device, "f", str(tmp_path / "f"), stamp=(11, 100)) + + assert device.rpc.read_requests[0].start_offset_bytes == 6 + assert (tmp_path / "f").read_bytes() == b"hello world" + assert sorted(p.name for p in tmp_path.iterdir()) == ["f"], "partial or stamp left behind" + + +def test_restarts_when_the_device_file_was_replaced(tmp_path): + # Same size, different modified time: a re-upload under the same name. + # Resuming would splice the old file's head onto the new file's tail. + _stamped_partial(tmp_path, b"OLDOLD", size=11, modified=100) + chunks = [ReadFileResponse(path="f", data=b"brand new!!", start_offset_bytes=0, file_size_bytes=11)] + device = _FakeDevice(read_chunks=chunks) + + files_client.read_file(device, "f", str(tmp_path / "f"), stamp=(11, 200)) + + assert device.rpc.read_requests[0].start_offset_bytes == 0 + assert (tmp_path / "f").read_bytes() == b"brand new!!" + + +def test_a_partial_with_no_stamp_is_restarted_when_a_stamp_is_given(tmp_path): + (tmp_path / "f.partial").write_bytes(b"who knows") + chunks = [ReadFileResponse(path="f", data=b"real", start_offset_bytes=0, file_size_bytes=4)] + device = _FakeDevice(read_chunks=chunks) + + files_client.read_file(device, "f", str(tmp_path / "f"), stamp=(4, 1)) + + assert device.rpc.read_requests[0].start_offset_bytes == 0 + assert (tmp_path / "f").read_bytes() == b"real" + + +def test_an_interrupted_download_records_which_version_it_holds(tmp_path): + import json + + def broken(): + yield ReadFileResponse(path="f", data=b"half", start_offset_bytes=0, file_size_bytes=8) + raise ConnectionError("link dropped") + + device = _FakeDevice() + device.rpc.ReadFile = lambda request: broken() + with pytest.raises(ConnectionError): + files_client.read_file(device, "f", str(tmp_path / "f"), stamp=(8, 42)) + + assert json.loads((tmp_path / "f.partial.json").read_text()) == {"size": 8, "modified": 42} diff --git a/synapse/utils/model_converter/pt_to_onnx.py b/synapse/utils/model_converter/pt_to_onnx.py index c7340ebe..c6a666ab 100644 --- a/synapse/utils/model_converter/pt_to_onnx.py +++ b/synapse/utils/model_converter/pt_to_onnx.py @@ -29,8 +29,12 @@ def convert_pt_to_onnx( import torch except ImportError: if console: - console.print("[bold red]Error:[/bold red] torch is required for PT to ONNX conversion") - console.print("[yellow]Install with: pip install torch[/yellow]") + console.print( + "[bold red]Error:[/bold red] .pt models need the model-conversion extra " + "(PyTorch and ONNX), which a plain install does not include." + ) + console.print("[yellow]Install it with: pip install 'science-synapse\\[model-convert]'[/yellow]") + console.print("[yellow]Or export your model to .onnx and deploy that instead.[/yellow]") return None if console: