diff --git a/.claude/CLAUDE.md b/.claude/CLAUDE.md index a13566d..5c75de5 100644 --- a/.claude/CLAUDE.md +++ b/.claude/CLAUDE.md @@ -6,6 +6,8 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co A Symfony 7.4 demo project showcasing [`cleverage/process-bundle`](https://github.com/cleverage/process-bundle) v5 and its bridge bundles (archive, cache, doctrine, flysystem, rest, soap, ui). The "product" here is not PHP code — it is the **process configurations** under `config/packages/process/`. Each file demonstrates one feature of the process ecosystem. `src/` contains only a handful of glue classes (entities, fixtures, a few custom adapters/DTOs). +> **Convention — how features are exercised here.** Features (including bug reproductions and regression checks) must be demonstrated **primarily through `clever_age_process` configurations**, each named with a `demo.` prefix and placed under `config/packages/process/` (one file per process, auto-imported). Reach for a throwaway PHP script or a custom `src/` command only as a last resort when a behaviour genuinely cannot be expressed as a process; prefer composing existing tasks and (generic) transformers instead. Example: issue-#24 SFTP reproduction lives in `config/packages/process/demo.sftp_stale_connection.yaml`. + ## Environment & commands Everything runs inside Docker (`.docker/compose.yaml`), driven by the `Makefile`. Do **not** run PHP/composer/console directly on the host — wrap them in the container. The Makefile targets already do this via `docker compose ... run --rm php-fpm`. diff --git a/.docker/compose.yaml b/.docker/compose.yaml index c038555..ebe45be 100644 --- a/.docker/compose.yaml +++ b/.docker/compose.yaml @@ -80,11 +80,35 @@ services: retries: 100 ports: - "${MYSQL_PORT:-3306}:3306" + sftp: <<: *base-sftp tty: true - ports: - - "${SFTP_PORT:-22}:22" + + # Same atmoz/sftp image as `sftp`, but with a deliberately short sshd idle + # timeout (see sftp/short-timeout.sh). Used by demo.sftp_stale_connection to + # reproduce flysystem-process-bundle issue #24; all other demo.sftp_* processes + # keep using the normal `sftp` service. + sftp-short: + <<: *base-sftp + tty: true + volumes: + - ./sftp/data:/home/sftp/data + - ./sftp/short-timeout.sh:/etc/sftp.d/short-timeout.sh:ro + + # Mock SFTP server (paramiko) that, after a short idle, injects a stray SFTP + # packet while keeping the SSH transport open - reproducing the client's SAP + # server behaviour and thus the VERBATIM issue #24 error + # ("Expected NET_SFTP_HANDLE or NET_SFTP_STATUS. Got packet type"). + sftp-mock: + build: + context: sftp-mock + tty: true + volumes: + - ./sftp/data:/data + environment: + SFTP_MOCK_ROOT: / + SFTP_MOCK_IDLE: "4" volumes: process_bundle_demo_data: diff --git a/.docker/sftp-mock/Dockerfile b/.docker/sftp-mock/Dockerfile new file mode 100644 index 0000000..7e3b585 --- /dev/null +++ b/.docker/sftp-mock/Dockerfile @@ -0,0 +1,8 @@ +FROM python:3.12-slim + +RUN pip install --no-cache-dir "paramiko>=3,<4" + +COPY server.py /server.py + +EXPOSE 22 +CMD ["python", "/server.py"] diff --git a/.docker/sftp-mock/server.py b/.docker/sftp-mock/server.py new file mode 100644 index 0000000..86c3729 --- /dev/null +++ b/.docker/sftp-mock/server.py @@ -0,0 +1,158 @@ +#!/usr/bin/env python3 +""" +Minimal mock SFTP server (paramiko) used to reproduce the EXACT flysystem-process-bundle +issue #24 error "Expected NET_SFTP_HANDLE or NET_SFTP_STATUS. Got packet type". + +It serves files normally, but after a short idle period on an established SFTP +session it injects one stray SFTP packet (SSH_FXP_NAME) onto the channel WHILE +KEEPING THE SSH TRANSPORT OPEN. This mimics enterprise SFTP servers (e.g. the +client's SAP server) that expire the SFTP session / leave a stray or late response +instead of tearing down the TCP transport the way OpenSSH does. phpseclib then reads +that stray packet on the next reused operation and raises the issue #24 error +(whereas a clean OpenSSH disconnect yields "Connection closed prematurely"). +""" +import os +import socket +import threading +import time + +import paramiko +from paramiko.sftp import CMD_NAME + +ROOT = os.environ.get("SFTP_MOCK_ROOT", "/data") +IDLE = float(os.environ.get("SFTP_MOCK_IDLE", "4")) +HOST_KEY = paramiko.RSAKey.generate(2048) + + +class Server(paramiko.ServerInterface): + def check_auth_password(self, username, password): + if username == "sftp" and password == "password": + return paramiko.AUTH_SUCCESSFUL + return paramiko.AUTH_FAILED + + def get_allowed_auths(self, username): + return "password" + + def check_channel_request(self, kind, chanid): + if kind == "session": + return paramiko.OPEN_SUCCEEDED + return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED + + +class StubSFTPHandle(paramiko.SFTPHandle): + def stat(self): + try: + return paramiko.SFTPAttributes.from_stat(os.fstat(self.readfile.fileno())) + except OSError as e: + return paramiko.SFTPServer.convert_errno(e.errno) + + +class StubSFTPServer(paramiko.SFTPServerInterface): + def _realpath(self, path): + return ROOT + self.canonicalize(path) + + def list_folder(self, path): + p = self._realpath(path) + out = [] + for fname in os.listdir(p): + attr = paramiko.SFTPAttributes.from_stat(os.stat(os.path.join(p, fname))) + attr.filename = fname + out.append(attr) + return out + + def stat(self, path): + try: + return paramiko.SFTPAttributes.from_stat(os.stat(self._realpath(path))) + except OSError as e: + return paramiko.SFTPServer.convert_errno(e.errno) + + def lstat(self, path): + try: + return paramiko.SFTPAttributes.from_stat(os.lstat(self._realpath(path))) + except OSError as e: + return paramiko.SFTPServer.convert_errno(e.errno) + + def open(self, path, flags, attr): + p = self._realpath(path) + try: + fd = os.open(p, flags, 0o666) + except OSError as e: + return paramiko.SFTPServer.convert_errno(e.errno) + if flags & os.O_WRONLY: + fstr = "ab" if flags & os.O_APPEND else "wb" + elif flags & os.O_RDWR: + fstr = "a+b" if flags & os.O_APPEND else "r+b" + else: + fstr = "rb" + try: + f = os.fdopen(fd, fstr) + except OSError as e: + return paramiko.SFTPServer.convert_errno(e.errno) + fobj = StubSFTPHandle(flags) + fobj.filename = p + fobj.readfile = f + fobj.writefile = f + return fobj + + +class InjectingSFTPServer(paramiko.SFTPServer): + """Serves normally, then injects one stray SFTP packet after IDLE seconds.""" + + def start_subsystem(self, name, transport, channel): + self._last = time.time() + self._seen = False + self._injected = False + self._alive = True + threading.Thread(target=self._watch, daemon=True).start() + try: + super().start_subsystem(name, transport, channel) + finally: + self._alive = False + + def _read_packet(self): + self._last = time.time() + self._seen = True + return super()._read_packet() + + def _watch(self): + while getattr(self, "_alive", False): + time.sleep(0.5) + if self._seen and not self._injected and (time.time() - self._last) > IDLE: + try: + # Stray SSH_FXP_NAME (type 104): body just needs valid framing; + # phpseclib reads the type, expects HANDLE/STATUS, and throws. + msg = paramiko.Message() + msg.add_int(0) # request id + msg.add_int(0) # count + self._send_packet(CMD_NAME, msg) + print("[sftp-mock] injected stray NAME packet after idle", flush=True) + except Exception as exc: # noqa: BLE001 + print(f"[sftp-mock] inject failed: {exc}", flush=True) + self._injected = True + + +def handle(client, addr): + try: + t = paramiko.Transport(client) + t.add_server_key(HOST_KEY) + t.set_subsystem_handler("sftp", InjectingSFTPServer, StubSFTPServer) + t.start_server(server=Server()) + while t.is_active(): + time.sleep(1) + except Exception as exc: # noqa: BLE001 + print(f"[sftp-mock] session error: {exc}", flush=True) + + +def main(): + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.bind(("", 22)) + sock.listen(10) + print(f"[sftp-mock] listening on :22 root={ROOT} idle={IDLE}s", flush=True) + while True: + client, addr = sock.accept() + threading.Thread(target=handle, args=(client, addr), daemon=True).start() + + +if __name__ == "__main__": + main() diff --git a/.docker/sftp/short-timeout.sh b/.docker/sftp/short-timeout.sh new file mode 100755 index 0000000..0cc8b77 --- /dev/null +++ b/.docker/sftp/short-timeout.sh @@ -0,0 +1,12 @@ +#!/bin/bash +# Shorten the sshd idle timeout for this dedicated SFTP service so that +# demo.sftp_stale_connection can reproduce flysystem-process-bundle issue #24 +# (stale cached SFTP connection) in a few seconds instead of the ~60 minutes +# reported in the wild. Run by the atmoz/sftp entrypoint (/etc/sftp.d/*) before +# sshd starts. Only this "short timeout" SFTP service mounts it; the normal +# `sftp` service keeps OpenSSH defaults. +set -e +{ + echo "ClientAliveInterval 3" + echo "ClientAliveCountMax 1" +} >> /etc/ssh/sshd_config diff --git a/.env b/.env index b28e0a2..d80163d 100644 --- a/.env +++ b/.env @@ -39,10 +39,14 @@ DATABASE_URL="mysql://app:app@mysql:3306/app?serverVersion=9.1.0&charset=utf8mb4 ###> docker/atmoz-sftp ### SFTP_HOST=sftp -SFTP_PORT=22 SFTP_USERNAME=sftp SFTP_PASSWORD=password SFTP_ROOT=data +# Dedicated SFTP service with a short sshd idle timeout (see .docker/sftp/short-timeout.sh). +SFTP_SHORT_HOST=sftp-short +# Mock SFTP server that injects a stray packet after idle (see .docker/sftp-mock/), to +# reproduce the verbatim issue #24 error. Serves /data as its root (SFTP root is empty). +SFTP_MOCK_HOST=sftp-mock ###< docker/atmoz-sftp ### ###> symfony/routing ### diff --git a/.gitignore b/.gitignore index 0b2f59e..3499663 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,7 @@ /composer.lock .idea .composer +.bash_history ###> symfony/framework-bundle ### /.env.local diff --git a/config/packages/cleverage_process.yaml b/config/packages/cleverage_process.yaml index 3d139f9..9635d58 100644 --- a/config/packages/cleverage_process.yaml +++ b/config/packages/cleverage_process.yaml @@ -31,4 +31,8 @@ clever_age_process: callback: callback: substr right_parameters: [ '{{ offset }}', '{{ length }}' ] + sleep: # pause for $value seconds (PHP sleep) then pass the value through + transformers: + callback: + callback: sleep diff --git a/config/packages/flysystem.yaml b/config/packages/flysystem.yaml index f70675c..9fd5b33 100644 --- a/config/packages/flysystem.yaml +++ b/config/packages/flysystem.yaml @@ -11,7 +11,57 @@ flysystem: adapter: 'sftp' options: host: '%env(string:SFTP_HOST)%' - port: '%env(int:SFTP_PORT)%' + port: 22 username: '%env(string:SFTP_USERNAME)%' password: '%env(string:SFTP_PASSWORD)%' - root: '%env(string:SFTP_ROOT)%' \ No newline at end of file + root: '%env(string:SFTP_ROOT)%' + + # `sftp-short` container (OpenSSH with a short sshd idle timeout). Two distinct + # storages (different roots) on the same host, following the .incoming / + # .archive convention. Because OpenSSH closes the idle transport cleanly, + # this yields "Connection closed prematurely". Used by demo.sftp_stale_connection. + remote.storage.short_timeout.incoming: + adapter: 'sftp' + options: + host: '%env(string:SFTP_SHORT_HOST)%' + port: 22 + username: '%env(string:SFTP_USERNAME)%' + password: '%env(string:SFTP_PASSWORD)%' + root: '%env(string:SFTP_ROOT)%/incoming' + connectivityChecker: App\Flysystem\SftpConnectivityChecker + + remote.storage.short_timeout.archive: + adapter: 'sftp' + options: + host: '%env(string:SFTP_SHORT_HOST)%' + port: 22 + username: '%env(string:SFTP_USERNAME)%' + password: '%env(string:SFTP_PASSWORD)%' + root: '%env(string:SFTP_ROOT)%/archive' + connectivityChecker: App\Flysystem\SftpConnectivityChecker + + # `sftp-mock` container: a paramiko server that injects a stray packet after idle + # while keeping the SSH transport open (like the client's SAP server). Two distinct + # storages (different roots) on the same host, following the .incoming / + # .archive convention. Being different storages, each gets its OWN + # SftpConnectionProvider (its OWN phpseclib connection). This yields the VERBATIM + # issue #24 error. Used by demo.sftp_fetch_archive_timeout. + remote.storage.mock.incoming: + adapter: 'sftp' + options: + host: '%env(string:SFTP_MOCK_HOST)%' + port: 22 + username: '%env(string:SFTP_USERNAME)%' + password: '%env(string:SFTP_PASSWORD)%' + root: '%env(string:SFTP_ROOT)%/incoming' + connectivityChecker: App\Flysystem\SftpConnectivityChecker + + remote.storage.mock.archive: + adapter: 'sftp' + options: + host: '%env(string:SFTP_MOCK_HOST)%' + port: 22 + username: '%env(string:SFTP_USERNAME)%' + password: '%env(string:SFTP_PASSWORD)%' + root: '%env(string:SFTP_ROOT)%/archive' + connectivityChecker: App\Flysystem\SftpConnectivityChecker diff --git a/config/packages/process/demo.sftp_fetch_archive_timeout.yaml b/config/packages/process/demo.sftp_fetch_archive_timeout.yaml new file mode 100644 index 0000000..dcb1488 --- /dev/null +++ b/config/packages/process/demo.sftp_fetch_archive_timeout.yaml @@ -0,0 +1,60 @@ +clever_age_process: + configurations: + demo.sftp_fetch_archive_timeout: + description: > + Minimal reproduction of the client's erp_contact_import failure (flysystem-process-bundle issue #24), + keeping only the essential shape: get_file fans out (in parallel) to [treatment, archive_file]. + Pass the get_file storage BASE via the `filesystem` context; get_file uses `{{ filesystem }}.incoming` + and archive_file uses `{{ filesystem }}.archive` (the archive storage is deduced from the base - option + templating is a plain substitution, so the suffix is appended). They are two DIFFERENT flysystem + storages (different roots) on the SAME host, so each has its OWN cached phpseclib connection; + archive_file opens a FRESH connection and does NOT reuse get_file's one. + The stale connection is get_file's own (`{{ filesystem }}.incoming`): opened by get_file.execute, left + idle during the treatment sleep, and re-hit by get_file.next() - FileFetchTask is iterable, so after the + outputs are processed it lists the source again on the now-stale connection and fails. This matches the + client report ("archiving task and subsequent file listing operations"). + With `filesystem: remote.storage.mock` the sftp-mock server injects a stray packet after idle while + keeping the transport open (like the client's SAP server), yielding the VERBATIM issue #24 error + "Expected NET_SFTP_HANDLE or NET_SFTP_STATUS. Got packet type". + help: > + bin/console cleverage:process:execute demo.sftp_fetch_archive_timeout -c filesystem:remote.storage.mock + tasks: + get_file: + service: '@CleverAge\FlysystemProcessBundle\Task\FileFetchTask' + description: > + Fetch the source file from `{{ filesystem }}.incoming` (opens + caches connection C_A). Mirrors + the client's get_zip; iterable (file_pattern), fans out to treatment and archive_file (LAST). + options: + source_filesystem: '{{ filesystem }}.incoming' + destination_filesystem: 'local.storage' + file_pattern: '/sample.*\.txt$/' + remove_source: false + ignore_missing: false + outputs: [treatment, debug, archive_file] + treatment: + service: '@CleverAge\ProcessBundle\Task\ConstantOutputTask' + description: > + Long-running treatment placeholder (feeds the idle duration in seconds to the sleep + transformer). Must exceed the server idle threshold. + options: + output: 10 + outputs: [treatment_sleep] + treatment_sleep: + service: '@CleverAge\ProcessBundle\Task\TransformerTask' + description: > + Keeps connection C_A idle (generic `sleep` transformer). Dead-end branch. + options: + transformers: + sleep: ~ + debug: + service: '@CleverAge\ProcessBundle\Task\Debug\DebugTask' + archive_file: + service: '@CleverAge\FlysystemProcessBundle\Task\FileFetchTask' + description: > + Archive the file by reading from `{{ filesystem }}.archive` - a DIFFERENT storage (deduced from + the base), so this opens a FRESH connection C_B. Mirrors the client's archive_file. + options: + source_filesystem: '{{ filesystem }}.incoming' + destination_filesystem: '{{ filesystem }}.archive' + remove_source: false + ignore_missing: false diff --git a/config/packages/process/demo.sftp_stale_connection.yaml b/config/packages/process/demo.sftp_stale_connection.yaml new file mode 100644 index 0000000..0cbf89a --- /dev/null +++ b/config/packages/process/demo.sftp_stale_connection.yaml @@ -0,0 +1,56 @@ +clever_age_process: + configurations: + demo.sftp_stale_connection: + description: > + Same shape as demo.sftp_fetch_archive_timeout (get_file -> treatment sleep -> archive_file, with + get_file re-listing its source on next()), but pointed at the OpenSSH `sftp-short` server via + `filesystem: remote.storage.short_timeout`. get_file uses `{{ filesystem }}.incoming` and archive_file + uses `{{ filesystem }}.archive` (archive storage deduced from the base). + This reproduces the ROOT CAUSE of flysystem-process-bundle issue #24 - a stale, cached SFTP connection + reused after the server timed out the idle session - but against OpenSSH the manifestation is + "Connection closed prematurely" (OpenSSH closes the idle transport cleanly with SSH_MSG_DISCONNECT, so + phpseclib hits EOF before reading a packet). For the VERBATIM issue-#24 wording + "Expected NET_SFTP_HANDLE or NET_SFTP_STATUS. Got packet type", run demo.sftp_fetch_archive_timeout + against `remote.storage.mock`, whose server leaves a stray packet on an open transport instead. + help: > + bin/console cleverage:process:execute demo.sftp_stale_connection -c filesystem:remote.storage.short_timeout + tasks: + get_file: + service: '@CleverAge\FlysystemProcessBundle\Task\FileFetchTask' + description: > + Fetch the source file from `{{ filesystem }}.incoming` (opens + caches the connection). + Iterable (file_pattern); fans out to treatment and archive_file (archive_file LAST). + options: + source_filesystem: '{{ filesystem }}.incoming' + destination_filesystem: 'local.storage' + file_pattern: '/sample.*\.txt$/' + remove_source: false + ignore_missing: false + outputs: [treatment, debug, archive_file] + treatment: + service: '@CleverAge\ProcessBundle\Task\ConstantOutputTask' + description: > + Long-running treatment placeholder (idle duration in seconds). Must exceed the server idle + threshold (sftp-short drops idle clients after ~6s). + options: + output: 10 + outputs: [treatment_sleep] + treatment_sleep: + service: '@CleverAge\ProcessBundle\Task\TransformerTask' + description: > + Keeps the cached connection idle (generic `sleep` transformer). Dead-end branch. + options: + transformers: + sleep: ~ + debug: + service: '@CleverAge\ProcessBundle\Task\Debug\DebugTask' + archive_file: + service: '@CleverAge\FlysystemProcessBundle\Task\FileFetchTask' + description: > + Archive by reading from `{{ filesystem }}.archive` - a DIFFERENT storage (deduced from the + base), so this opens a FRESH connection. + options: + source_filesystem: '{{ filesystem }}.incoming' + destination_filesystem: '{{ filesystem }}.archive' + remove_source: false + ignore_missing: false diff --git a/src/Flysystem/SftpConnectivityChecker.php b/src/Flysystem/SftpConnectivityChecker.php new file mode 100644 index 0000000..f002b05 --- /dev/null +++ b/src/Flysystem/SftpConnectivityChecker.php @@ -0,0 +1,40 @@ +isConnected()) { + return false; + } + + try { + // A transport-level ping() (SSH2::ping opens a separate keep-alive + // channel) cannot detect a de-synchronized SFTP stream: if the server + // left a stray/late packet on the SFTP channel, ping() still succeeds. + // So probe at the SFTP protocol level - if the stream is desynced this + // round-trip reads the stray packet and throws, and we force a reconnect. + $connection->stat('.'); + + return true; + } catch (\Throwable) { + return false; + } + } +}