mjalmoudy94
72408a0a4f
The one path that touches a customer's server, and the one the platform cannot undo. It is human-triggered, confirmed and audited; there is deliberately no code path from a pipeline run into it (D13). - deploy_ssh.py: SshDeployer. Release lands in releases/<ts>-<sha7>/, the post-deploy command runs with cwd there, and only then does `current` move — atomically (relative symlink + mv -T). A failed hook or extraction never moves it, so the previous release keeps serving. Host key is pinned on first success (TOFU); a mismatch is a hard failure. Prunes to the newest five. - deploy_ftp.py: FtpDeployer. FTP has no atomic switch and no rollback except a backup, so a backup is mandatory: every file about to be overwritten is archived under ARTIFACTS_ROOT before anything is uploaded. The artifact is extracted, then the build dir (dist/build/public/out) is chosen — a static deploy must not publish the repository. Archive members are sanitised: absolute paths, `..` and escaping links are refused. - deploy_target.py: the tester that dispatches on `kind`. - deploy_service.py + production_deploys (migration a3b4c5d6e7f4) + 4 routes. One deploy per project at a time; the org slug must be typed back; the target summary reports has_secret, never the secret. Tests: 499 API (+26) with fake SSH/FTP transports; ruff, mypy, alembic clean. Report: docs/plan/reports/agent-3-phase-5.md
476 lines
18 KiB
Python
476 lines
18 KiB
Python
"""SSH / SFTP production deploy (D8).
|
|
|
|
Production is the one path in the product that a human must trigger, and the one
|
|
that can destroy a running site. Three rules follow from that and shape
|
|
everything here.
|
|
|
|
**Nothing is uploaded into the live directory.** A release lands in
|
|
``<remote_path>/releases/<timestamp>-<sha7>/`` and is switched in by moving a
|
|
``current`` symlink. That makes the switch atomic (``ln -sfn`` + ``mv`` on the
|
|
same filesystem) and makes rollback a pointer move rather than a re-upload.
|
|
|
|
**The host key is pinned on first success (TOFU).** The first ``test_connection``
|
|
that succeeds records the fingerprint; every later connection must match it. A
|
|
changed key is a hard failure, not a warning — it is either a rebuilt server (in
|
|
which case the operator re-tests deliberately) or an interception.
|
|
|
|
**Credentials never touch the target's disk.** The password or key is held in
|
|
memory for the connection only; the deploy scripts that run on the server are
|
|
built from a fixed template and never interpolate a secret.
|
|
|
|
The transport is injectable so the deploy logic — release numbering, pruning,
|
|
the symlink dance, rollback — is testable without a live sshd.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import logging
|
|
import os
|
|
import shlex
|
|
import time
|
|
from collections.abc import Awaitable, Callable
|
|
from dataclasses import dataclass
|
|
from typing import Any, Protocol, runtime_checkable
|
|
|
|
from app.application.ports.deploy_target import DeployCheck, DeployTargetSpec
|
|
from app.core.redact import redact_text
|
|
from app.domain.platform.rules import assert_remote_host_allowed
|
|
|
|
logger = logging.getLogger("agy.deploy.ssh")
|
|
|
|
CONNECT_TIMEOUT_SECONDS = 15.0
|
|
COMMAND_TIMEOUT_SECONDS = 600.0
|
|
PROBE_TIMEOUT_SECONDS = 20.0
|
|
|
|
#: How many releases to keep on the target. Five is enough to roll back past a
|
|
#: bad deploy and a bad deploy of its fix, without filling the operator's disk.
|
|
KEEP_RELEASES = 5
|
|
|
|
#: Name of the pointer the web server is configured to serve from.
|
|
CURRENT_LINK = "current"
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class CommandOutcome:
|
|
exit_code: int
|
|
output: str = ""
|
|
timed_out: bool = False
|
|
|
|
@property
|
|
def ok(self) -> bool:
|
|
return self.exit_code == 0 and not self.timed_out
|
|
|
|
|
|
@runtime_checkable
|
|
class SshTransport(Protocol):
|
|
"""The remote operations a deploy needs. Implemented over asyncssh."""
|
|
|
|
async def connect(
|
|
self,
|
|
*,
|
|
host: str,
|
|
port: int,
|
|
username: str,
|
|
password: str | None,
|
|
key: str | None,
|
|
known_hosts: str | None,
|
|
) -> None: ...
|
|
|
|
def host_key_fingerprint(self) -> str: ...
|
|
|
|
async def run(
|
|
self, command: str, *, timeout: float = COMMAND_TIMEOUT_SECONDS
|
|
) -> CommandOutcome: ...
|
|
|
|
async def upload(self, *, local_path: str, remote_path: str) -> None: ...
|
|
|
|
async def close(self) -> None: ...
|
|
|
|
|
|
TransportFactory = Callable[[], SshTransport]
|
|
|
|
|
|
class AsyncSshTransport:
|
|
"""The real transport: one asyncssh connection per deploy."""
|
|
|
|
def __init__(self) -> None:
|
|
# asyncssh is imported lazily inside connect(); the attribute is typed
|
|
# loosely so this module imports without the dependency at hand.
|
|
self._conn: Any = None
|
|
self._fingerprint = ""
|
|
|
|
async def connect(
|
|
self,
|
|
*,
|
|
host: str,
|
|
port: int,
|
|
username: str,
|
|
password: str | None,
|
|
key: str | None,
|
|
known_hosts: str | None,
|
|
) -> None:
|
|
import asyncssh
|
|
|
|
# `known_hosts=None` disables verification for this connection; the
|
|
# fingerprint is captured and compared by the caller, which is where the
|
|
# TOFU policy lives. Passing an explicit `known_hosts` string enforces
|
|
# the pinned key inside asyncssh as well.
|
|
kwargs: dict = {
|
|
"host": host,
|
|
"port": port,
|
|
"username": username,
|
|
"connect_timeout": CONNECT_TIMEOUT_SECONDS,
|
|
"known_hosts": known_hosts if known_hosts else None,
|
|
}
|
|
if key:
|
|
kwargs["client_keys"] = [asyncssh.import_private_key(key)]
|
|
kwargs["password"] = None
|
|
else:
|
|
kwargs["password"] = password or ""
|
|
self._conn = await asyncssh.connect(**kwargs)
|
|
self._fingerprint = self._read_fingerprint()
|
|
|
|
def _read_fingerprint(self) -> str:
|
|
"""Best-effort host key digest of the connection we just made."""
|
|
try:
|
|
key = self._conn.get_server_host_key()
|
|
if key is None:
|
|
return ""
|
|
digest = key.get_fingerprint("sha256")
|
|
return f"SHA256:{digest}" if not str(digest).startswith("SHA256:") else str(digest)
|
|
except Exception: # noqa: BLE001 - a missing fingerprint must not block a deploy
|
|
return ""
|
|
|
|
def host_key_fingerprint(self) -> str:
|
|
return self._fingerprint
|
|
|
|
async def run(
|
|
self, command: str, *, timeout: float = COMMAND_TIMEOUT_SECONDS
|
|
) -> CommandOutcome:
|
|
import asyncssh
|
|
|
|
assert self._conn is not None, "connect() first"
|
|
try:
|
|
result = await self._conn.run(command, check=False, timeout=timeout)
|
|
except (asyncssh.Error, TimeoutError) as exc:
|
|
return CommandOutcome(exit_code=1, output=redact_text(str(exc))[:2000], timed_out=True)
|
|
output = f"{result.stdout or ''}{result.stderr or ''}"
|
|
return CommandOutcome(exit_code=int(result.exit_status or 0), output=output)
|
|
|
|
async def upload(self, *, local_path: str, remote_path: str) -> None:
|
|
assert self._conn is not None, "connect() first"
|
|
async with self._conn.start_sftp_client() as sftp:
|
|
await sftp.put(local_path, remote_path)
|
|
|
|
async def close(self) -> None:
|
|
if self._conn is not None:
|
|
self._conn.close()
|
|
with contextlib.suppress(Exception):
|
|
await self._conn.wait_closed()
|
|
self._conn = None
|
|
|
|
|
|
class SshDeployer:
|
|
"""``DeployTargetTester`` and the production deploy itself, over SSH."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
transport_factory: TransportFactory = AsyncSshTransport,
|
|
artifacts_root: str | None = None,
|
|
allow_private_hosts: bool = False,
|
|
clock: Callable[[], float] = time.time,
|
|
) -> None:
|
|
from app.core.config import settings
|
|
|
|
self._factory = transport_factory
|
|
self.artifacts_root = artifacts_root or settings.artifacts_root
|
|
self.allow_private_hosts = allow_private_hosts
|
|
self._clock = clock
|
|
|
|
# -- testing -------------------------------------------------------------
|
|
|
|
async def test_connection(self, *, target: DeployTargetSpec) -> DeployCheck:
|
|
"""Connect, pin the host key, and confirm the path is writable."""
|
|
try:
|
|
assert_remote_host_allowed(
|
|
f"ssh://{target.host}:{target.port}", allow_private=self.allow_private_hosts
|
|
)
|
|
except Exception as exc: # noqa: BLE001 - translated to a plain message
|
|
return DeployCheck(ok=False, message=str(exc))
|
|
|
|
transport = self._factory()
|
|
try:
|
|
await transport.connect(
|
|
host=target.host,
|
|
port=target.port,
|
|
username=target.username,
|
|
password=target.secret if target.auth_type == "password" else None,
|
|
key=target.secret if target.auth_type == "key" else None,
|
|
known_hosts=None,
|
|
)
|
|
except Exception as exc: # noqa: BLE001 - never surface the raw error
|
|
return DeployCheck(ok=False, message=_ssh_error_message(exc))
|
|
finally:
|
|
with contextlib.suppress(Exception):
|
|
await transport.close()
|
|
|
|
fingerprint = transport.host_key_fingerprint()
|
|
if target.known_fingerprint and fingerprint and fingerprint != target.known_fingerprint:
|
|
# A changed host key is a hard failure: it is either a rebuilt server
|
|
# or an interception, and only a human can tell the two apart.
|
|
return DeployCheck(
|
|
ok=False,
|
|
message="کلید میزبان تغییر کرده است؛ برای ادامه باید دوباره تأیید کنید.",
|
|
fingerprint=fingerprint,
|
|
)
|
|
|
|
probe = await _probe_writable(transport, target.remote_path)
|
|
if not probe.ok:
|
|
return DeployCheck(
|
|
ok=False,
|
|
message=f"مسیر «{target.remote_path}» قابل نوشتن نیست.",
|
|
fingerprint=fingerprint,
|
|
)
|
|
return DeployCheck(
|
|
ok=True, message="اتصال و مسیر مقصد تأیید شد.", fingerprint=fingerprint
|
|
)
|
|
|
|
# -- deploy --------------------------------------------------------------
|
|
|
|
async def deploy(
|
|
self,
|
|
*,
|
|
target: DeployTargetSpec,
|
|
artifact_path: str,
|
|
sha: str,
|
|
post_deploy_command: str | None = None,
|
|
on_output: Callable[[str], Awaitable[None]] | None = None,
|
|
) -> tuple[bool, str, str]:
|
|
"""Upload, extract, run the post-deploy command, switch ``current``.
|
|
|
|
Returns ``(ok, release_dir, message)``. On failure the ``current``
|
|
symlink is untouched, so a failed deploy leaves the previous release
|
|
serving — the same property the demo pipeline has.
|
|
"""
|
|
if not os.path.isfile(artifact_path):
|
|
return False, "", "فایل بستهٔ انتشار پیدا نشد."
|
|
|
|
transport = self._factory()
|
|
try:
|
|
await transport.connect(
|
|
host=target.host,
|
|
port=target.port,
|
|
username=target.username,
|
|
password=target.secret if target.auth_type == "password" else None,
|
|
key=target.secret if target.auth_type == "key" else None,
|
|
known_hosts=None,
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
return False, "", _ssh_error_message(exc)
|
|
|
|
try:
|
|
fingerprint = transport.host_key_fingerprint()
|
|
if target.known_fingerprint and fingerprint and fingerprint != target.known_fingerprint:
|
|
return False, "", "کلید میزبان تغییر کرده است."
|
|
|
|
release = _release_name(self._clock(), sha)
|
|
release_dir = f"{target.remote_path.rstrip('/')}/releases/{release}"
|
|
|
|
setup = await transport.run(
|
|
f"mkdir -p {shlex.quote(release_dir)} && "
|
|
f"mkdir -p {shlex.quote(target.remote_path.rstrip('/') + '/releases')}",
|
|
timeout=PROBE_TIMEOUT_SECONDS,
|
|
)
|
|
if not setup.ok:
|
|
return False, "", "ساخت پوشهٔ انتشار در سرور مقصد ناموفق بود."
|
|
|
|
archive = f"{release_dir}/release.tar.gz"
|
|
await transport.upload(local_path=artifact_path, remote_path=archive)
|
|
await _stream(on_output, f"[deploy] بسته در {archive} بارگذاری شد.\n")
|
|
|
|
extract = await transport.run(
|
|
f"cd {shlex.quote(release_dir)} && tar -xzf release.tar.gz && rm -f release.tar.gz",
|
|
timeout=COMMAND_TIMEOUT_SECONDS,
|
|
)
|
|
if on_output is not None:
|
|
await on_output(redact_text(extract.output))
|
|
if not extract.ok:
|
|
return False, release_dir, "استخراج بسته در سرور مقصد ناموفق بود."
|
|
|
|
if post_deploy_command:
|
|
post = await transport.run(
|
|
f"cd {shlex.quote(release_dir)} && {post_deploy_command}",
|
|
timeout=COMMAND_TIMEOUT_SECONDS,
|
|
)
|
|
if on_output is not None:
|
|
await on_output(redact_text(post.output))
|
|
if not post.ok:
|
|
# The previous release keeps serving: `current` never moved.
|
|
return False, release_dir, "اجرای دستور پس از استقرار ناموفق بود."
|
|
|
|
switch = await transport.run(
|
|
_switch_script(target.remote_path, release),
|
|
timeout=PROBE_TIMEOUT_SECONDS,
|
|
)
|
|
if not switch.ok:
|
|
return False, release_dir, "تغییر نسخهٔ فعال ناموفق بود."
|
|
|
|
prune = await transport.run(
|
|
_prune_script(target.remote_path, KEEP_RELEASES),
|
|
timeout=PROBE_TIMEOUT_SECONDS,
|
|
)
|
|
if not prune.ok:
|
|
# Pruning is housekeeping; a full disk is not a failed deploy.
|
|
logger.warning("ssh deploy: pruning old releases failed: %s", prune.output[:200])
|
|
|
|
return True, release_dir, "استقرار در production انجام شد."
|
|
except Exception as exc: # noqa: BLE001 - never leak the raw error
|
|
logger.exception("ssh deploy failed")
|
|
return False, "", _ssh_error_message(exc)
|
|
finally:
|
|
with contextlib.suppress(Exception):
|
|
await transport.close()
|
|
|
|
async def rollback(
|
|
self,
|
|
*,
|
|
target: DeployTargetSpec,
|
|
to_release: str | None = None,
|
|
post_deploy_command: str | None = None,
|
|
on_output: Callable[[str], Awaitable[None]] | None = None,
|
|
) -> tuple[bool, str, str]:
|
|
"""Re-point ``current`` at the previous release and re-run the hook."""
|
|
transport = self._factory()
|
|
try:
|
|
await transport.connect(
|
|
host=target.host,
|
|
port=target.port,
|
|
username=target.username,
|
|
password=target.secret if target.auth_type == "password" else None,
|
|
key=target.secret if target.auth_type == "key" else None,
|
|
known_hosts=None,
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
return False, "", _ssh_error_message(exc)
|
|
|
|
try:
|
|
releases = await transport.run(
|
|
_list_releases_script(target.remote_path), timeout=PROBE_TIMEOUT_SECONDS
|
|
)
|
|
if not releases.ok:
|
|
return False, "", "خواندن فهرست نسخهها ناموفق بود."
|
|
names = [line.strip() for line in releases.output.splitlines() if line.strip()]
|
|
if len(names) < 2:
|
|
return False, "", "نسخهٔ قبلیای برای بازگردانی وجود ندارد."
|
|
|
|
# `ls -1t` is newest first, so the previous release is the second.
|
|
current = names[0]
|
|
target_name = to_release or names[1]
|
|
if target_name == current:
|
|
target_name = names[1]
|
|
|
|
switch = await transport.run(
|
|
_switch_script(target.remote_path, target_name), timeout=PROBE_TIMEOUT_SECONDS
|
|
)
|
|
if not switch.ok:
|
|
return False, "", "بازگردانی ناموفق بود."
|
|
|
|
if post_deploy_command:
|
|
post = await transport.run(
|
|
f"cd {shlex.quote(target.remote_path.rstrip('/') + '/releases/' + target_name)}"
|
|
f" && {post_deploy_command}",
|
|
timeout=COMMAND_TIMEOUT_SECONDS,
|
|
)
|
|
if on_output is not None:
|
|
await on_output(redact_text(post.output))
|
|
if not post.ok:
|
|
return False, target_name, "دستور پس از بازگردانی ناموفق بود."
|
|
|
|
return True, target_name, f"بازگردانی به نسخهٔ {target_name} انجام شد."
|
|
except Exception as exc: # noqa: BLE001
|
|
return False, "", _ssh_error_message(exc)
|
|
finally:
|
|
with contextlib.suppress(Exception):
|
|
await transport.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# remote scripts — built from constants, never from user input
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _release_name(now: float, sha: str) -> str:
|
|
stamp = time.strftime("%Y%m%dT%H%M%S", time.gmtime(now))
|
|
return f"{stamp}-{(sha or 'unknown')[:7]}"
|
|
|
|
|
|
def _switch_script(remote_path: str, release: str) -> str:
|
|
"""Move the ``current`` symlink atomically.
|
|
|
|
A relative symlink into ``releases/`` plus ``mv -T`` on the same filesystem
|
|
is the only switch that is atomic *and* survives the directory being moved.
|
|
"""
|
|
base = shlex.quote(remote_path.rstrip("/"))
|
|
rel = shlex.quote(f"releases/{release}")
|
|
link = shlex.quote(f"{remote_path.rstrip('/')}/{CURRENT_LINK}")
|
|
tmp = shlex.quote(f"{remote_path.rstrip('/')}/.{CURRENT_LINK}.new")
|
|
return f"ln -sfn {rel} {tmp} && mv -T {tmp} {link} && cd {base} && ls -l {link}"
|
|
|
|
|
|
def _prune_script(remote_path: str, keep: int) -> str:
|
|
"""Delete all but the newest ``keep`` release directories."""
|
|
releases = shlex.quote(f"{remote_path.rstrip('/')}/releases")
|
|
return (
|
|
f"cd {releases} && ls -1t | tail -n +{keep + 1} | "
|
|
f"while read -r old; do rm -rf -- \"$old\"; done"
|
|
)
|
|
|
|
|
|
def _list_releases_script(remote_path: str) -> str:
|
|
releases = shlex.quote(f"{remote_path.rstrip('/')}/releases")
|
|
return f"cd {releases} && ls -1t"
|
|
|
|
|
|
async def _probe_writable(transport: SshTransport, remote_path: str) -> CommandOutcome:
|
|
base = shlex.quote(remote_path.rstrip("/") or "/")
|
|
return await transport.run(
|
|
f"mkdir -p {base} && test -w {base} && "
|
|
f"probe={base}/.smart-org-probe && touch $probe && rm -f $probe",
|
|
timeout=PROBE_TIMEOUT_SECONDS,
|
|
)
|
|
|
|
|
|
async def _stream(callback: Callable[[str], Awaitable[None]] | None, text: str) -> None:
|
|
if callback is not None:
|
|
await callback(text)
|
|
|
|
|
|
def _ssh_error_message(exc: BaseException) -> str:
|
|
"""A plain Persian message. Never the raw exception, which can carry the host."""
|
|
text = str(exc).lower()
|
|
if "permission denied" in text or "auth" in text:
|
|
return "احراز هویت روی سرور مقصد ناموفق بود."
|
|
if "timed out" in text or "timeout" in text:
|
|
return "اتصال به سرور مقصد در زمان مجاز برقرار نشد."
|
|
if "host key" in text:
|
|
return "کلید میزبان تأیید نشد."
|
|
if "no such file" in text or "not found" in text:
|
|
return "مسیر یا فایل مقصد پیدا نشد."
|
|
if "connection refused" in text or "unreachable" in text:
|
|
return "اتصال به سرور مقصد برقرار نشد."
|
|
return "استقرار در سرور مقصد ناموفق بود."
|
|
|
|
|
|
__all__ = [
|
|
"COMMAND_TIMEOUT_SECONDS",
|
|
"CONNECT_TIMEOUT_SECONDS",
|
|
"CURRENT_LINK",
|
|
"KEEP_RELEASES",
|
|
"AsyncSshTransport",
|
|
"CommandOutcome",
|
|
"SshDeployer",
|
|
"SshTransport",
|
|
]
|