Support Warpgate SSH selectors and interactive authentication

README.md

@@ -32,6 +32,10 @@ WebSSH keeps terminal work, files, commands, diagnostics, and notes in one
 responsive browser workspace. It is self-hosted, multi-user, and built without
 a hosted control plane or runtime CDN dependencies.
 
+After an administrator enables the optional gateway integration in Settings → Integrations,
+connect through an existing SSH gateway with a user:target username. Disabled by default. See the
+[Warpgate integration guide](docs/warpgate.md) for authentication and target requirements.
+
 ## Why WebSSH
 
 - **One workspace, not a terminal tab.** Keep SSH sessions, SFTP sources,

app/__init__.py

@@ -22,7 +22,8 @@
     get_user_settings,
     save_user_settings,
 )
-from .app_settings import is_registration_enabled, set_registration_enabled
+from .app_settings import (is_registration_enabled, set_registration_enabled,
+                           is_ssh_gateway_enabled, set_ssh_gateway_enabled)
 from .storage_errors import StorageCorruptionError
 from .tailscale_ssh import user_can_use_tailscale_ssh
 from .runtime_lifecycle import RuntimeLifecycle
@@ -465,6 +466,7 @@ def inject_url_prefix():
         return {
             'url_prefix': url_prefix,
             'registration_enabled': registration_available,
+            'ssh_gateway_enabled': is_ssh_gateway_enabled(),
             'tmux_enabled': config.TMUX_ENABLED,
             'tmux_default': config.TMUX_DEFAULT,
             'admin_panel_enabled': config.ADMIN_PANEL_ENABLED,
@@ -725,6 +727,12 @@ def enforce_security_feature_gate():
                 transfer_runtime_binding
             ),
         )
+    from .ssh_connection_attempt import SSHAttemptRegistry
+    ssh_attempts = SSHAttemptRegistry()
+    app.extensions['ssh_attempt_registry'] = ssh_attempts
+    app.extensions['runtime_lifecycle'].register_shutdown_callback(
+        'ssh_attempts', lambda _deadline: ssh_attempts.shutdown(),
+    )
     cors_origins = config.CORS_ORIGINS
     if isinstance(cors_origins, str):
         cors_origins = [origin.strip() for origin in cors_origins.split(',') if origin.strip()]
@@ -1485,7 +1493,8 @@ def admin_audit():
     @admin_required
     @login_required
     def admin_get_settings():
-        return jsonify({'registration_enabled': is_registration_enabled()})
+        return jsonify({'registration_enabled': is_registration_enabled(),
+                        'ssh_gateway_enabled': is_ssh_gateway_enabled()})
 
     @app.route('/admin/api/settings', methods=['POST'])
     @admin_required
@@ -1495,6 +1504,9 @@ def admin_set_settings():
         data = request.get_json(silent=True)
         if not isinstance(data, dict):
             return jsonify({'error': 'Invalid settings payload'}), 400
+        if ('ssh_gateway_enabled' in data
+                and type(data['ssh_gateway_enabled']) is not bool):
+            return jsonify({'error': 'ssh_gateway_enabled must be a boolean'}), 400
         if 'registration_enabled' in data:
             if type(data['registration_enabled']) is not bool:
                 return jsonify({
@@ -1513,7 +1525,12 @@ def admin_set_settings():
             val = set_registration_enabled(data['registration_enabled'])
             log_info("Admin changed registration setting",
                      admin=current_user.username, registration_enabled=val)
-        return jsonify({'registration_enabled': is_registration_enabled()})
+        if 'ssh_gateway_enabled' in data:
+            val = set_ssh_gateway_enabled(data['ssh_gateway_enabled'])
+            log_info("Admin changed SSH gateway setting",
+                     admin=current_user.username, ssh_gateway_enabled=val)
+        return jsonify({'registration_enabled': is_registration_enabled(),
+                        'ssh_gateway_enabled': is_ssh_gateway_enabled()})
 
     @app.route('/admin/api/security-features', methods=['GET'])
     @admin_required

app/app_settings.py

@@ -16,6 +16,10 @@ def _valid_settings(value):
     return (
         isinstance(value, dict)
         and value.get('schema_version') == CURRENT_STORAGE_VERSIONS['app_settings']
+        and (
+            'ssh_gateway_enabled' not in value
+            or type(value['ssh_gateway_enabled']) is bool
+        )
         and (
             'registration_enabled' not in value
             or type(value['registration_enabled']) is bool
@@ -82,6 +86,23 @@ def get_audit_backup_count():
     ))
 
 
+def is_ssh_gateway_enabled():
+    """Gateway authentication requires an explicit, persisted admin opt-in."""
+    return _load().get('ssh_gateway_enabled') is True
+
+
+def set_ssh_gateway_enabled(value):
+    if type(value) is not bool:
+        return False
+    with storage_lock(f'app-settings:{_SETTINGS_FILE}'):
+        data = _load_with_lock_held()
+        data['ssh_gateway_enabled'] = value
+        if not _valid_settings(data):
+            return False
+        _save(data)
+    return value
+
+
 def set_audit_backup_count(value):
     if type(value) is not int or not 1 <= value <= 90:
         return False

app/connection_pool.py

@@ -68,7 +68,7 @@ def bind_lifecycle(self, lifecycle):
             )
             return self.cleanup_handle
 
-    def create_connection(self, host, port, username, password=None, key_path=None, key_content=None, user_id=None):
+    def create_connection(self, host, port, username, password=None, key_path=None, key_content=None, user_id=None, gateway_attempt=None):
         """
         Create a temporary SSH+SFTP connection.
 
@@ -132,7 +132,21 @@ def create_connection(self, host, port, username, password=None, key_path=None,
                 'allow_agent': False
             }
 
-            if key_content:
+            if gateway_attempt is not None:
+                from .ssh_gateway_auth import GatewayAuthStrategy, GatewayTransport
+                gateway_attempt.own(client)
+                gateway_attempt.own(validated_socket)
+                if key_path:
+                    return None, "Gateway authentication requires a stored key"
+                connect_kwargs.pop('look_for_keys', None)
+                connect_kwargs.pop('allow_agent', None)
+                connect_kwargs['transport_factory'] = GatewayTransport
+                connect_kwargs['auth_strategy'] = GatewayAuthStrategy(
+                    username, password=password,
+                    pkey=_load_private_key(key_content) if key_content else None,
+                    interact=gateway_attempt.challenge, check=gateway_attempt.check,
+                )
+            elif key_content:
                 connect_kwargs['pkey'] = _load_private_key(key_content)
             elif key_path:
                 connect_kwargs['key_filename'] = key_path
@@ -149,11 +163,18 @@ def create_connection(self, host, port, username, password=None, key_path=None,
             if transport:
                 transport.set_keepalive(30)
 
-            sftp = open_sftp_client(
-                transport,
-                timeout=config.SSH_CONNECT_TIMEOUT,
-                operation_timeout=config.SFTP_OPERATION_TIMEOUT,
-            )
+            if gateway_attempt is not None:
+                from .ssh_gateway_setup import prepare_sftp
+                sftp = prepare_sftp(
+                    transport, gateway_attempt,
+                    operation_timeout=config.SFTP_OPERATION_TIMEOUT,
+                )
+            else:
+                sftp = open_sftp_client(
+                    transport,
+                    timeout=config.SSH_CONNECT_TIMEOUT,
+                    operation_timeout=config.SFTP_OPERATION_TIMEOUT,
+                )
 
             conn_id = uuid.uuid4().hex
 
@@ -164,6 +185,8 @@ def create_connection(self, host, port, username, password=None, key_path=None,
                     and not lifecycle.accepting_work()
                 ):
                     return None, "Runtime is shutting down"
+                if gateway_attempt is not None:
+                    gateway_attempt.handoff(client, validated_socket, sftp)
                 self.connections[conn_id] = {
                     'client': client,
                     'sftp': sftp,

app/profile_manager.py

@@ -354,9 +354,21 @@ def _validate_profile_payload(user_id, payload, dependent_lock_held=False):
     except (ValueError, TypeError):
         return None, 'Invalid port number'
 
-    username = str(username).strip()
-    if not re.match(r'^[a-zA-Z0-9_\-\.]{1,32}$', username):
-        return None, 'Invalid username format'
+    from .ssh_gateway import parse_selector
+    if isinstance(username, str) and ':' in username:
+        from .app_settings import is_ssh_gateway_enabled
+        if not is_ssh_gateway_enabled():
+            return None, 'SSH gateway integration is disabled by the administrator'
+        if auth_type == 'tailscale':
+            return None, 'Gateway selectors cannot use Tailscale SSH'
+        try:
+            parse_selector(username)
+        except ValueError as error:
+            return None, str(error)
+    else:
+        username = str(username).strip()
+        if not re.match(r'^[a-zA-Z0-9_\-\.]{1,32}$', username):
+            return None, 'Invalid username format'
     if auth_type not in {'password', 'key', 'tailscale'}:
         return None, 'Invalid auth_type'
     if auth_type == 'tailscale' and payload.get('jump_host_id'):

app/sftp_handler.py

@@ -1,3 +1,4 @@
+import errno
 import hashlib
 import os
 import socket
@@ -227,6 +228,43 @@ class RemoteMetadataLimitExceeded(SFTPOperationError):
 
 def public_sftp_error(error, fallback=_PUBLIC_SFTP_ERROR):
     """Return only small, application-authored SFTP errors to clients."""
+    code = getattr(error, 'errno', None)
+    if isinstance(error, PermissionError) or code in (errno.EACCES, errno.EPERM):
+        return 'Permission denied'
+    if isinstance(error, FileNotFoundError) or code == errno.ENOENT:
+        return 'File or directory not found'
+    if isinstance(error, NotADirectoryError) or code == errno.ENOTDIR:
+        return 'Not a directory'
+    if isinstance(error, FileExistsError) or code == errno.EEXIST:
+        return 'File or directory already exists'
+    if code == errno.EROFS:
+        return 'Remote file system is read-only'
+    if code == errno.ENOSPC:
+        return 'Remote file system is full'
+    if isinstance(error, (socket.timeout, TimeoutError)) or code == errno.ETIMEDOUT:
+        return 'Remote file operation timed out'
+
+    if isinstance(error, (OSError, SFTPError, SFTPOperationError)):
+        message = str(error).strip().lower()
+        if len(message) <= _PUBLIC_SFTP_ERROR_MAX_BYTES:
+            if message in {'permission denied', 'access denied',
+                           'operation not permitted'} or any(
+                message.startswith(reason + ': ')
+                for reason in ('permission denied', 'access denied',
+                               'operation not permitted')
+            ):
+                return 'Permission denied'
+            if message in {'no such file', 'no such file or directory'}:
+                return 'File or directory not found'
+            if message == 'not a directory':
+                return 'Not a directory'
+            if message == 'file exists':
+                return 'File or directory already exists'
+            if message == 'read-only file system':
+                return 'Remote file system is read-only'
+            if message == 'no space left on device':
+                return 'Remote file system is full'
+
     if isinstance(error, SFTPOperationError):
         message = str(error)
         if len(message) > _PUBLIC_SFTP_ERROR_MAX_BYTES: