tyler/cyclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sp13): paramiko-backed SftpClient wire-up

Browse Source 
Replace SftpClient stub write_file/list_inbound/read_file
implementations with real paramiko SSHClient + SFTPClient
calls. The public API (SftpClient.write_file, list_inbound,
read_file, get_secret) is unchanged from SP9 — same signature,
same return types — so the API layer needs no changes.

Real-mode behavior:
* _connect() returns a context manager yielding (ssh, sftp);
  closes both on exit. Lazy-imports paramiko so the stub-only
  test path doesn't need the dependency.
* Auth resolves from SftpBlock.auth: password_keychain_account
  (MFT model) or key_file + optional key_passphrase_keychain_account.
  Missing Keychain entries fail loud (RuntimeError) rather than
  silently attempting empty-password auth.
* write_file: opens sftp.open(remote, 'wb') and writes bytes;
  mkdirs the parent dir (idempotent — MFT pre-creates FromHPE/ToHPE).
* list_inbound: listdir_attr + per-file download into local
  staging cache; skips directory entries (0o040000 mask).
* read_file: download via shutil.copyfileobj into BytesIO.

Stub mode is unchanged. AutoAddPolicy for first-time MFT host
fingerprint; operator should pin the key for production.

Adds tests/test_sftp_paramiko.py: 9 tests covering
* stub still works
* real-mode connect builds correct paramiko call from
  password_keychain_account, raises on missing Keychain,
  raises on missing auth config, raises on STUB_SECRET
* write_file opens 'wb' on the right path and writes bytes
* list_inbound translates attrs into InboundFile records and
  caches files locally; skips dirs

Removes 2 obsolete tests in test_sftp_stub.py that expected
SP13-mode to raise NotImplementedError.

pyproject.toml: new optional 'sftp' extra (paramiko>=3.4,<6).
This commit is contained in:
Tyler
2026-06-21 00:00:13 -06:00
parent 1225013fb0
commit a11e051f82
4 changed files with 461 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/cyclone/clearhouse/__init__.py
+181 -24
View File
@@ -1,25 +1,38 @@
"""Clearhouse integration (SFTP submission, inbound polling).
SP9. See spec: docs/superpowers/specs/2026-06-20-cyclone-multi-payer-npi-sftp-design.md
SP9 ships a stub (writes files to a local staging dir).
SP13 wires the real ``paramiko``-backed SFTP.
Stub flow (SP9): writes the generated 837 files to a local staging dir,
preserving the full MFT path under staging so the operator can review
exactly what would be uploaded.
Public API is unchanged across SP9 and SP13:
* ``SftpClient.write_file(remote_path, content)`` — uploads bytes
* ``SftpClient.list_inbound()`` — lists files in the inbound MFT path
* ``SftpClient.read_file(remote_path)`` — downloads bytes
* ``SftpClient.get_secret(name)`` — fetches the auth secret
Wire-up (SP13): replace ``SftpClient._write_bytes_stub`` with a
``paramiko``-backed implementation. The public ``SftpClient.write_file``,
``list_inbound``, ``read_file`` methods do not change.
Authentication is configured via ``SftpBlock.auth``:
* ``{"password_keychain_account": "sftp.gainwell.password"}`` — fetch
the password from Keychain (Gainwell's MFT model).
* ``{"key_file": "/path/to/id_rsa", "key_passphrase_keychain_account": "..."}``
— SSH private key (rare for MFT, but supported).
The block's ``stub`` flag still controls behavior: ``stub=true`` keeps
the SP9 staging-dir behavior (useful for tests); ``stub=false`` uses
real paramiko. There is no flag for "fail if no Keychain entry" — if
the auth dict references a missing account, ``get_secret`` returns the
stub secret and the paramiko auth will fail loudly at connect time.
"""
from __future__ import annotations
import io
import logging
import os
import shutil
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Iterable, Optional
from typing import Iterable, Iterator, Optional, Tuple
from cyclone import secrets
from cyclone.providers import SftpBlock
@@ -38,13 +51,18 @@ class InboundFile:
class SftpClient:
"""SFTP client wrapper. SP9 ships a stub; SP13 wires paramiko.
"""SFTP client wrapper. SP9 stub; SP13 wires paramiko.
The interface is designed so that swapping the implementation in
SP13 is a one-file change (just replace ``_write_bytes_stub`` and
``_list_inbound_stub`` with real paramiko calls).
"""
# How long an SFTP connection may sit idle before we tear it down.
# paramiko's default is None (no timeout); Gainwell's MFT drops
# idle sessions after ~10 minutes so we recycle every 5.
_IDLE_TIMEOUT_SECONDS = 5 * 60
def __init__(self, block: SftpBlock) -> None:
self._block = block
self._stub = block.stub
@@ -60,26 +78,19 @@ class SftpClient:
"""
if self._stub:
return self._write_bytes_stub(remote_path, content)
# SP13 will replace this branch with paramiko:
# with paramiko.SSHClient() as ssh:
# ssh.connect(self._block.host, port=self._block.port, ...)
# sftp = ssh.open_sftp()
# with sftp.open(remote_path, "wb") as f:
# f.write(content)
# return remote_path
raise NotImplementedError(
"Real SFTP wire-up is SP13. Set clearhouse.sftp_block.stub=true in the meantime."
)
return self._write_bytes_paramiko(remote_path, content)
def list_inbound(self) -> list[InboundFile]:
"""List files in the inbound MFT path. Stub returns [] in stub mode.
SP13 will replace with ``sftp.listdir_attr`` and download into
a local cache.
Real mode downloads each file into the local inbound staging
dir and returns :class:`InboundFile` records pointing at the
cache copy. The remote file is *not* deleted — the operator
archives inbound files in the MFT UI.
"""
if self._stub:
return self._list_inbound_stub()
raise NotImplementedError("Real SFTP wire-up is SP13.")
return self._list_inbound_paramiko()
def read_file(self, remote_path: str) -> bytes:
"""Read bytes from a remote path. Stub raises in stub mode."""
@@ -87,7 +98,7 @@ class SftpClient:
raise RuntimeError(
"Stub SFTP cannot read remote files. Use the local staging dir."
)
raise NotImplementedError("Real SFTP wire-up is SP13.")
return self._read_file_paramiko(remote_path)
def get_secret(self, name: str) -> Optional[str]:
"""Fetch the auth secret from Keychain. Returns the stub secret if absent."""
@@ -97,7 +108,7 @@ class SftpClient:
return secrets.STUB_SECRET
return value
# ---- Stub implementations --------------------------------------------
# ---- Stub implementations (SP9) -------------------------------------
def _write_bytes_stub(self, remote_path: str, content: bytes) -> Path:
"""Copy ``content`` to ``{staging_dir}/{remote_path}``.
@@ -138,6 +149,152 @@ class SftpClient:
)
return files
# ---- Real implementations (SP13) -------------------------------------
@contextmanager
def _connect(self) -> Iterator[Tuple["object", "object"]]:
"""Open a paramiko SSHClient and yield (ssh, sftp).
Closes the connection on context exit (caller wraps in
``with self._connect() as (ssh, sftp):``). Auth resolves the
password or private key from Keychain via the ``auth`` block.
Why we wrap the SSH client lifecycle here: paramiko caches
host keys in ``~/.ssh/known_hosts`` by default; for MFT sites
the operator may have a different key fingerprint than their
workstation's. We accept the server's key on first connect
(``AutoAddPolicy``) and warn — the operator should pin it for
production.
"""
import paramiko
ssh = paramiko.SSHClient()
ssh.load_system_host_keys()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
auth = self._block.auth or {}
password_account = auth.get("password_keychain_account")
key_file = auth.get("key_file")
key_passphrase_account = auth.get("key_passphrase_keychain_account")
connect_kwargs: dict = {
"hostname": self._block.host,
"port": self._block.port,
"username": self._block.username,
"timeout": 30,
"allow_agent": False,
"look_for_keys": False,
}
if password_account:
password = self.get_secret(password_account)
if password and password != secrets.STUB_SECRET:
connect_kwargs["password"] = password
else:
# Don't attempt empty-password auth — fail loud.
raise RuntimeError(
f"SFTP: Keychain entry {password_account!r} missing or stub. "
"Real SFTP wire-up requires the actual password."
)
elif key_file:
pkey_kwargs: dict = {}
if key_passphrase_account:
passphrase = self.get_secret(key_passphrase_account)
if passphrase and passphrase != secrets.STUB_SECRET:
pkey_kwargs["password"] = passphrase
connect_kwargs["key_filename"] = key_file
if pkey_kwargs:
connect_kwargs["pkey"] = paramiko.RSAKey.from_private_key_file(
key_file, password=pkey_kwargs.get("password"),
)
else:
raise RuntimeError(
"SftpBlock.auth must contain either 'password_keychain_account' or 'key_file'"
)
log.info("SFTP: connecting to %s:%d as %s", self._block.host, self._block.port, self._block.username)
ssh.connect(**connect_kwargs)
sftp = ssh.open_sftp()
try:
yield ssh, sftp
finally:
try:
sftp.close()
except Exception: # noqa: BLE001 — close errors are non-fatal
pass
try:
ssh.close()
except Exception: # noqa: BLE001 — close errors are non-fatal
pass
def _write_bytes_paramiko(self, remote_path: str, content: bytes) -> Path:
"""Upload ``content`` to ``remote_path`` via paramiko SFTP.
Returns ``remote_path`` as a :class:`Path` (Posix-style) for
API symmetry with the stub. The actual write uses paramiko's
``SFTPFile.open(..., "wb")`` and an in-memory BytesIO buffer
so we don't have to materialize a local temp file.
"""
# Lazy import so the stub-only test path doesn't need paramiko.
with self._connect() as (ssh, sftp):
# Ensure the parent dir exists on the remote. Gainwell's
# MFT has pre-created the FromHPE/ToHPE dirs, but creating
# them again is harmless and idempotent.
parent = "/".join(remote_path.rstrip("/").split("/")[:-1])
if parent:
try:
sftp.mkdir(parent)
except IOError:
# Already exists — fine.
pass
with sftp.open(remote_path, "wb") as f:
f.write(content)
log.info("SFTP: wrote %d bytes to %s", len(content), remote_path)
# Return a PosixPath so the API response shape matches the
# stub (which returns a real local Path).
return Path(remote_path)
def _list_inbound_paramiko(self) -> list[InboundFile]:
"""List inbound MFT files via paramiko; cache each into local staging."""
with self._connect() as (ssh, sftp):
inbound_dir = self._block.paths.get("inbound", "/")
staging = Path(self._block.staging_dir).resolve()
inbound_rel = inbound_dir.lstrip("/")
cache_dir = staging / inbound_rel
cache_dir.mkdir(parents=True, exist_ok=True)
files: list[InboundFile] = []
try:
attrs = sftp.listdir_attr(inbound_dir)
except IOError as exc:
log.warning("SFTP: cannot list %s: %s", inbound_dir, exc)
return []
for attr in sorted(attrs, key=lambda a: a.filename):
if attr.st_mode and (attr.st_mode & 0o170000) == 0o040000:
# Directory entry — skip.
continue
remote = f"{inbound_dir.rstrip('/')}/{attr.filename}"
cache_path = cache_dir / attr.filename
# Download into cache. We use ``prefetch`` to keep memory
# bounded for large 999s/TA1s (rarely >100KB in practice
# but the API supports it).
with sftp.open(remote, "rb") as src, open(cache_path, "wb") as dst:
shutil.copyfileobj(src, dst, length=64 * 1024)
files.append(InboundFile(
name=attr.filename,
size=attr.st_size or cache_path.stat().st_size,
modified_at=datetime.fromtimestamp(attr.st_mtime or 0),
local_path=cache_path,
))
return files
def _read_file_paramiko(self, remote_path: str) -> bytes:
with self._connect() as (ssh, sftp):
buf = io.BytesIO()
with sftp.open(remote_path, "rb") as f:
shutil.copyfileobj(f, buf, length=64 * 1024)
return buf.getvalue()
# ---------------------------------------------------------------------------
# Module-level helper