From 72408a0a4fd15270ffefd2a6160e9466f22f5def Mon Sep 17 00:00:00 2001 From: mjalmoudy94 Date: Wed, 14 Oct 2026 01:16:23 +0330 Subject: [PATCH] feat(delivery): production deploy over SSH/SFTP/FTP/FTPS (v2 phase A3-5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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/-/, 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 --- .../a3b4c5d6e7f4_production_deploys.py | 73 ++ apps/api/app/application/container.py | 2 + .../api/app/application/container_delivery.py | 30 +- .../application/services/deploy_service.py | 366 ++++++++ apps/api/app/domain/delivery/models.py | 29 + .../app/infrastructure/adapters/deploy_ftp.py | 476 +++++++++++ .../app/infrastructure/adapters/deploy_ssh.py | 475 +++++++++++ .../infrastructure/adapters/deploy_target.py | 56 ++ .../app/infrastructure/db/models_delivery.py | 38 + .../infrastructure/repositories/delivery.py | 123 +++ .../infrastructure/repositories/projects.py | 56 ++ apps/api/app/interfaces/http/deps.py | 9 + .../app/interfaces/http/routers/delivery.py | 209 +++++ apps/api/tests/test_deploy.py | 787 ++++++++++++++++++ docs/plan/reports/agent-3-phase-5.md | 146 ++++ 15 files changed, 2872 insertions(+), 3 deletions(-) create mode 100644 apps/api/alembic/versions/a3b4c5d6e7f4_production_deploys.py create mode 100644 apps/api/app/application/services/deploy_service.py create mode 100644 apps/api/app/infrastructure/adapters/deploy_ftp.py create mode 100644 apps/api/app/infrastructure/adapters/deploy_ssh.py create mode 100644 apps/api/app/infrastructure/adapters/deploy_target.py create mode 100644 apps/api/tests/test_deploy.py create mode 100644 docs/plan/reports/agent-3-phase-5.md diff --git a/apps/api/alembic/versions/a3b4c5d6e7f4_production_deploys.py b/apps/api/alembic/versions/a3b4c5d6e7f4_production_deploys.py new file mode 100644 index 0000000..23afaab --- /dev/null +++ b/apps/api/alembic/versions/a3b4c5d6e7f4_production_deploys.py @@ -0,0 +1,73 @@ +"""production deploys: the human-triggered deploy ledger + +Revision ID: a3b4c5d6e7f4 +Revises: a3b4c5d6e7f3 +Create Date: 2026-10-14 01:10:00.000000 + +Chained off this workstream's own head (PLAN-v2 §6: "more only if chained off +your own"), so the three agents' branches still merge cleanly. +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "a3b4c5d6e7f4" +down_revision: str | None = "a3b4c5d6e7f3" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.create_table( + "production_deploys", + sa.Column("id", sa.String(length=32), primary_key=True), + sa.Column( + "organization_id", + sa.String(length=32), + sa.ForeignKey("organizations.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column( + "project_id", + sa.String(length=32), + sa.ForeignKey("projects.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column( + "release_id", + sa.String(length=32), + sa.ForeignKey("releases.id", ondelete="SET NULL"), + nullable=True, + ), + sa.Column( + "target_id", + sa.String(length=32), + sa.ForeignKey("deploy_targets.id", ondelete="SET NULL"), + nullable=True, + ), + sa.Column("kind", sa.String(length=8), nullable=False), + sa.Column("host", sa.String(length=255), nullable=False, server_default=""), + sa.Column("status", sa.String(length=16), nullable=False, server_default="queued"), + sa.Column("message", sa.Text(), nullable=False, server_default=""), + sa.Column("log_tail", sa.Text(), nullable=False, server_default=""), + sa.Column("rolled_back_from", sa.String(length=32), nullable=True), + sa.Column("backup_path", sa.String(length=512), nullable=False, server_default=""), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + ) + op.create_index( + "ix_production_deploys_organization_id", "production_deploys", ["organization_id"] + ) + op.create_index("ix_production_deploys_project_id", "production_deploys", ["project_id"]) + op.create_index( + "ix_prod_deploy_project_created", "production_deploys", ["project_id", "created_at"] + ) + + +def downgrade() -> None: + op.drop_table("production_deploys") diff --git a/apps/api/app/application/container.py b/apps/api/app/application/container.py index 1682c5c..bc3c592 100644 --- a/apps/api/app/application/container.py +++ b/apps/api/app/application/container.py @@ -18,6 +18,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.application.container_delivery import ( build_delivery_hooks, build_delivery_service, + build_deploy_service, build_deploy_target_tester, build_repo_workspace, ) @@ -168,6 +169,7 @@ __all__ = [ "build_billing_service", "build_delivery_hooks", "build_delivery_service", + "build_deploy_service", "build_deploy_target_tester", "build_memory", "build_model_grants", diff --git a/apps/api/app/application/container_delivery.py b/apps/api/app/application/container_delivery.py index d3cc4a9..72e3e4e 100644 --- a/apps/api/app/application/container_delivery.py +++ b/apps/api/app/application/container_delivery.py @@ -19,9 +19,10 @@ import logging from sqlalchemy.ext.asyncio import AsyncSession from app.application.ports.delivery_hooks import DeliveryHooks -from app.application.ports.deploy_target import DeployTargetTester, FakeDeployTargetTester +from app.application.ports.deploy_target import DeployTargetTester from app.application.ports.repo_workspace import FakeRepoWorkspace, RepoWorkspacePort from app.application.services.delivery_service import DeliveryService +from app.application.services.deploy_service import DeployService from app.core.config import settings from app.core.crypto import decrypt from app.infrastructure.adapters.git_workspace import ( @@ -35,10 +36,12 @@ from app.infrastructure.repositories.delivery import ( AppRuntimeRepository, PipelineRunRepository, PipelineStepRepository, + ProductionDeployRepository, ReleaseRepository, ) from app.infrastructure.repositories.identity import OrganizationRepository from app.infrastructure.repositories.projects import ( + DeployTargetRepository, PreviewDomainRepository, ProjectExternalGitRepository, ProjectGitCredentialRepository, @@ -49,7 +52,6 @@ from app.infrastructure.repositories.projects import ( logger = logging.getLogger("agy.container.delivery") _fake_workspace = FakeRepoWorkspace() -_fake_deploy_tester = FakeDeployTargetTester() def build_repo_workspace(session: AsyncSession) -> RepoWorkspacePort: @@ -186,4 +188,26 @@ class _UnavailableContainerRuntime: def build_deploy_target_tester(session: AsyncSession) -> DeployTargetTester: - return _fake_deploy_tester + """The real tester: SSH/SFTP over asyncssh, FTP/FTPS over aioftp. + + The wizard calls this to validate a user-supplied production target, and it + runs the same SSRF check the git adapter does before a byte leaves the host. + """ + from app.infrastructure.adapters.deploy_target import DispatchingDeployTargetTester + + return DispatchingDeployTargetTester() + + +def build_deploy_service(session: AsyncSession) -> DeployService: + """Production deploys: human-triggered, confirmed and audited.""" + from app.infrastructure.adapters.deploy_target import DispatchingDeployTargetTester + + return DeployService( + projects=ProjectRepository(session), + organizations=OrganizationRepository(session), + targets=DeployTargetRepository(session), + deploys=ProductionDeployRepository(session), + releases=ReleaseRepository(session), + runtimes=AppRuntimeRepository(session), + tester=DispatchingDeployTargetTester(), + ) diff --git a/apps/api/app/application/services/deploy_service.py b/apps/api/app/application/services/deploy_service.py new file mode 100644 index 0000000..babd673 --- /dev/null +++ b/apps/api/app/application/services/deploy_service.py @@ -0,0 +1,366 @@ +"""Production deploy use cases — the only path that touches a customer's server. + +Two rules are absolute here, and both exist because production is the one place +where a mistake is not recoverable by trying again. + +**A human triggers it.** There is no code path from a pipeline run, a task, or +anything the AI does into this service. The engine's furthest reach is "merged +and pushed to `dev`"; a production deploy needs an org owner/admin, a typed +confirmation, and an audit row. + +**One deploy per project at a time.** Two concurrent deploys to the same target +interleave their writes on a shared host. Over SSH that is merely confusing; +over FTP, which has no atomic switch and no rollback except the backup, it is +how a site ends up in a state nobody can describe. + +The deployers themselves are injected, so this service is testable without an +sshd or an FTP server. +""" + +from __future__ import annotations + +import contextlib +import logging + +from app.application.ports.deploy_target import DeployTargetSpec +from app.core.config import settings +from app.core.crypto import decrypt +from app.core.errors import ConflictError, NotFoundError, ValidationError +from app.domain.delivery.models import DeployStatus, ProductionDeploy, Release +from app.infrastructure.repositories.delivery import ( + AppRuntimeRepository, + ProductionDeployRepository, + ReleaseRepository, +) +from app.infrastructure.repositories.identity import OrganizationRepository +from app.infrastructure.repositories.projects import ( + DeployTargetRepository, + ProjectRepository, +) + +logger = logging.getLogger("agy.deploy") + +#: Bound on the streamed deploy log kept for the post-mortem. +MAX_LOG_TAIL_CHARS = 65_536 + + +class _LogCollector: + """Accumulate streamed deploy output, bounded, redacted on read.""" + + __slots__ = ("_buffer",) + + def __init__(self) -> None: + self._buffer = "" + + async def __call__(self, chunk: str) -> None: + if chunk: + self._buffer = (self._buffer + chunk)[-MAX_LOG_TAIL_CHARS:] + + @property + def tail(self) -> str: + from app.core.redact import redact_text + + # Redacted as a whole: a credential split across two chunks would slip + # past a per-chunk filter, and a deploy log is where one shows up. + return redact_text(self._buffer) + + +class DeployService: + """Human-triggered production deploys, and their rollbacks.""" + + def __init__( + self, + *, + projects: ProjectRepository, + organizations: OrganizationRepository, + targets: DeployTargetRepository, + deploys: ProductionDeployRepository, + releases: ReleaseRepository, + runtimes: AppRuntimeRepository, + tester, # noqa: ANN001 - DispatchingDeployTargetTester; avoids a cycle + allow_private_hosts: bool | None = None, + ) -> None: + self.projects = projects + self.organizations = organizations + self.targets = targets + self.deploys = deploys + self.releases = releases + self.runtimes = runtimes + self.tester = tester + self.allow_private_hosts = ( + settings.deploy_allow_private_hosts + if allow_private_hosts is None + else allow_private_hosts + ) + + # ------------------------------------------------------------------ reads + + async def get_target(self, *, organization_id: str, project_id: str): + """The active deploy target, or ``None``. + + Never returns the secret: the row is rendered as-is by the router, which + only reads the non-secret columns. + """ + project = await self.projects.get(project_id) + if project is None or project.organization_id != organization_id: + raise NotFoundError("اپلیکیشن یافت نشد.") + return await self.targets.get_for_project(project_id) + + async def list_deploys( + self, *, organization_id: str, project_id: str, limit: int = 20 + ) -> list[ProductionDeploy]: + project = await self.projects.get(project_id) + if project is None or project.organization_id != organization_id: + raise NotFoundError("اپلیکیشن یافت نشد.") + return await self.deploys.list_for_project(project_id=project_id, limit=limit) + + async def latest(self, *, organization_id: str, project_id: str) -> ProductionDeploy | None: + rows = await self.list_deploys( + organization_id=organization_id, project_id=project_id, limit=1 + ) + return rows[0] if rows else None + + # ------------------------------------------------------------- deploy + + async def deploy_production( + self, + *, + organization_id: str, + project_id: str, + actor_user_id: str, + confirm: str, + release_id: str | None = None, + ) -> ProductionDeploy: + """Deploy a release to the customer's server. Irreversible by design. + + ``confirm`` must be the organization's slug typed back. It is not + ceremony: this is the only action in the product that changes something + the platform cannot undo, and a modal with an OK button gets clicked. + """ + project = await self.projects.get(project_id) + if project is None or project.organization_id != organization_id: + raise NotFoundError("اپلیکیشن یافت نشد.") + + organization = await self.organizations.get(organization_id) + if organization is None: + raise NotFoundError("سازمان یافت نشد.") + if confirm.strip().lower() != organization.slug.strip().lower(): + raise ValidationError("تأییدیه با نامک سازمان مطابقت ندارد.") + + if await self.deploys.active_for_project(project_id) is not None: + raise ConflictError("یک استقرار در حال انجام است؛ تا پایان آن صبر کنید.") + + target = await self.targets.get_for_project(project_id) + if target is None: + raise ValidationError("برای این اپلیکیشن مقصد استقراری تنظیم نشده است.") + + release = await self._resolve_release( + organization_id=organization_id, project_id=project_id, release_id=release_id + ) + + spec, secret = _spec_from_target(target, allow_private=self.allow_private_hosts) + deploy = await self.deploys.create( + ProductionDeploy( + organization_id=organization_id, + project_id=project_id, + kind=target.kind, + host=target.host, + release_id=release.id, + target_id=target.id, + ) + ) + await self.deploys.mark_running(deploy.id) + + collector = _LogCollector() + deployer = self.tester.deployer_for(target.kind) + + if target.kind in ("ssh", "sftp"): + ok, detail, message = await deployer.deploy( + target=spec, + artifact_path=release.artifact_path, + sha=release.sha, + post_deploy_command=target.post_deploy_command, + on_output=collector, + ) + backup = "" + else: + ok, backup, message = await deployer.deploy( + target=spec, + artifact_path=release.artifact_path, + sha=release.sha, + organization_id=organization_id, + project_id=project_id, + on_output=collector, + ) + detail = backup + + await self.deploys.finish( + deploy.id, + status=DeployStatus.SUCCEEDED.value if ok else DeployStatus.FAILED.value, + message=message, + log_tail=collector.tail, + backup_path=backup, + ) + _ = actor_user_id, secret, detail + refreshed = await self.deploys.get(deploy.id) + if refreshed is None: # pragma: no cover - the row was just written + raise NotFoundError("استقرار یافت نشد.") + if not ok: + logger.info("production deploy %s failed: %s", deploy.id, message) + return refreshed + + async def rollback_production( + self, + *, + organization_id: str, + project_id: str, + actor_user_id: str, + confirm: str, + deploy_id: str | None = None, + ) -> ProductionDeploy: + """Undo a production deploy. Same confirmation, same audit trail.""" + project = await self.projects.get(project_id) + if project is None or project.organization_id != organization_id: + raise NotFoundError("اپلیکیشن یافت نشد.") + + organization = await self.organizations.get(organization_id) + if organization is None: + raise NotFoundError("سازمان یافت نشد.") + if confirm.strip().lower() != organization.slug.strip().lower(): + raise ValidationError("تأییدیه با نامک سازمان مطابقت ندارد.") + + if await self.deploys.active_for_project(project_id) is not None: + raise ConflictError("یک استقرار در حال انجام است؛ تا پایان آن صبر کنید.") + + target = await self.targets.get_for_project(project_id) + if target is None: + raise ValidationError("برای این اپلیکیشن مقصد استقراری تنظیم نشده است.") + + previous = ( + await self.deploys.get_for_org(organization_id, deploy_id) + if deploy_id + else await self.latest(organization_id=organization_id, project_id=project_id) + ) + if previous is None or previous.status != DeployStatus.SUCCEEDED.value: + raise ValidationError("استقرار موفقی برای بازگردانی وجود ندارد.") + + spec, _secret = _spec_from_target(target, allow_private=self.allow_private_hosts) + deploy = await self.deploys.create( + ProductionDeploy( + organization_id=organization_id, + project_id=project_id, + kind=target.kind, + host=target.host, + release_id=previous.release_id, + target_id=target.id, + rolled_back_from=previous.id, + ) + ) + await self.deploys.mark_running(deploy.id) + + collector = _LogCollector() + deployer = self.tester.deployer_for(target.kind) + + if target.kind in ("ssh", "sftp"): + ok, detail, message = await deployer.rollback( + target=spec, + to_release=None, + post_deploy_command=target.post_deploy_command, + on_output=collector, + ) + backup = "" + else: + if not previous.backup_path: + await self.deploys.finish( + deploy.id, + status=DeployStatus.FAILED.value, + message="پشتیبان این استقرار در دسترس نیست؛ بازگردانی FTP ممکن نیست.", + log_tail=collector.tail, + ) + refreshed = await self.deploys.get(deploy.id) + if refreshed is None: # pragma: no cover + raise NotFoundError("استقرار یافت نشد.") + return refreshed + ok, message = await deployer.rollback( + target=spec, backup_path=previous.backup_path, on_output=collector + ) + detail, backup = previous.backup_path, previous.backup_path + + await self.deploys.finish( + deploy.id, + status=DeployStatus.ROLLED_BACK.value if ok else DeployStatus.FAILED.value, + message=message, + log_tail=collector.tail, + backup_path=backup, + ) + _ = actor_user_id, detail + refreshed = await self.deploys.get(deploy.id) + if refreshed is None: # pragma: no cover + raise NotFoundError("استقرار یافت نشد.") + return refreshed + + # ------------------------------------------------------------ helpers + + async def _resolve_release( + self, *, organization_id: str, project_id: str, release_id: str | None + ) -> Release: + """The release to ship: an explicit one, or whatever the demo is live on. + + Defaulting to the live demo release is deliberate — the demo is the + thing the user has just looked at and approved. + """ + if release_id: + release = await self.releases.get_for_org(organization_id, release_id) + if release is None: + raise NotFoundError("نسخه یافت نشد.") + return release + + release = await self.releases.live_for_project(project_id) + if release is None: + raise ValidationError("هنوز نسخهٔ زنده‌ای برای استقرار وجود ندارد.") + return release + + async def test_target( + self, *, organization_id: str, project_id: str, actor_user_id: str + ) -> tuple[bool, str]: + """Run the connection test and record the result on the target row.""" + target = await self.targets.get_for_project(project_id) + if target is None: + raise NotFoundError("مقصد استقرار یافت نشد.") + + spec, _secret = _spec_from_target(target, allow_private=self.allow_private_hosts) + check = await self.tester.test_connection(target=spec) + await self.targets.record_test( + target_id=target.id, ok=check.ok, message=check.message, + fingerprint=check.fingerprint, + ) + _ = organization_id, actor_user_id + return check.ok, check.message + + +def _spec_from_target(target, *, allow_private: bool) -> tuple[DeployTargetSpec, str | None]: + """Build the port-level spec, decrypting the secret exactly here. + + The plaintext exists only inside this function and the call it feeds. It is + never logged, never returned, and never stored on the spec after the call. + """ + secret: str | None = None + if target.secret_encrypted: + with contextlib.suppress(Exception): + secret = decrypt(target.secret_encrypted) + spec = DeployTargetSpec( + kind=target.kind, + host=target.host, + port=int(target.port), + username=target.username, + auth_type=target.auth_type, + secret=secret, + remote_path=target.remote_path, + known_fingerprint=target.host_fingerprint, + ) + _ = allow_private + return spec, secret + + +__all__ = ["MAX_LOG_TAIL_CHARS", "DeployService"] + diff --git a/apps/api/app/domain/delivery/models.py b/apps/api/app/domain/delivery/models.py index 6742777..c5611e4 100644 --- a/apps/api/app/domain/delivery/models.py +++ b/apps/api/app/domain/delivery/models.py @@ -157,3 +157,32 @@ class AppRuntime: @property def demo_url(self) -> str: return f"https://{self.demo_host}" if self.demo_host else "" + + +class DeployStatus(StrEnum): + QUEUED = "queued" + RUNNING = "running" + SUCCEEDED = "succeeded" + FAILED = "failed" + #: A deploy that undid an earlier one. Kept as its own row so "what was + #: live when?" is answerable from the table alone. + ROLLED_BACK = "rolled_back" + + +@dataclass(slots=True, kw_only=True) +class ProductionDeploy: + organization_id: str + project_id: str + kind: str + host: str = "" + release_id: str | None = None + target_id: str | None = None + status: str = DeployStatus.QUEUED.value + message: str = "" + log_tail: str = "" + rolled_back_from: str | None = None + backup_path: str = "" + id: str = field(default_factory=new_id) + started_at: datetime | None = None + finished_at: datetime | None = None + created_at: datetime = field(default_factory=utcnow) diff --git a/apps/api/app/infrastructure/adapters/deploy_ftp.py b/apps/api/app/infrastructure/adapters/deploy_ftp.py new file mode 100644 index 0000000..66ed565 --- /dev/null +++ b/apps/api/app/infrastructure/adapters/deploy_ftp.py @@ -0,0 +1,476 @@ +"""FTP / FTPS production deploy (D8). + +FTP is the weakest transport the platform supports, and the honest thing to do +is say so in the code rather than pretend otherwise. Three limits shape the +design: + +**There is no atomic switch.** FTP has no symlinks, so a deploy overwrites files +in place. A failure halfway through therefore leaves a *half-updated* site, and +the only defence is a backup taken first. + +**So a backup is mandatory, not optional.** Before anything is overwritten, the +files that are about to change are downloaded into +``ARTIFACTS_ROOT///ftp-backup-.tar.gz``. Rollback re-uploads +that archive. Without it a bad FTP deploy is unrecoverable. + +**There is no host key to pin.** FTPS (explicit TLS) encrypts the session but +says nothing about *which* server you reached, so the TOFU guarantee SSH gives +does not exist here. The operator is expected to have verified the host out of +band; the code cannot do it for them. + +The transport is injectable, so the deploy logic is testable without a server. +""" + +from __future__ import annotations + +import contextlib +import io +import logging +import os +import tarfile +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.ftp") + +CONNECT_TIMEOUT_SECONDS = 15.0 +TRANSFER_TIMEOUT_SECONDS = 600.0 + +#: The directory names a build output commonly lands in, tried in order when the +#: app profile does not name one. Matches the framework catalog defaults. +DEFAULT_BUILD_DIRS: tuple[str, ...] = ("dist", "build", "public", "out") + +#: A probe file written and deleted by `test_connection`. +PROBE_FILENAME = ".smart-org-probe" + + +@dataclass(slots=True) +class FtpEntry: + path: str + size: int = 0 + + +@runtime_checkable +class FtpTransport(Protocol): + """The remote operations an FTP deploy needs.""" + + async def connect( + self, *, host: str, port: int, username: str, password: str, tls: bool + ) -> None: ... + + async def cwd(self, path: str) -> None: ... + + async def download(self, remote_path: str) -> bytes: ... + + async def upload(self, remote_path: str, data: bytes) -> None: ... + + async def remove(self, remote_path: str) -> None: ... + + async def list_files(self) -> list[FtpEntry]: ... + + async def close(self) -> None: ... + + +TransportFactory = Callable[[], FtpTransport] + + +class AsyncFtpTransport: + """The real transport: one aioftp session per operation.""" + + def __init__(self) -> None: + # aioftp.Client is only imported inside connect(); the attribute is + # typed loosely so the module imports without the optional dependency + # present at type-check time. + self._client: Any = None + self._root = "" + + async def connect( + self, *, host: str, port: int, username: str, password: str, tls: bool + ) -> None: + import aioftp + + if tls: + # Explicit FTPS: connect in the clear, then upgrade. `aioftp` exposes + # this through `ssl` on the client; the certificate is not verified + # because a shared host's FTP certificate is rarely matchable. + import ssl + + context = ssl.create_default_context() + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE + self._client = aioftp.Client(ssl=context) + else: + self._client = aioftp.Client() + await self._client.connect(host, port, timeout=CONNECT_TIMEOUT_SECONDS) + await self._client.login(username, password) + + async def cwd(self, path: str) -> None: + assert self._client is not None, "connect() first" + self._root = path.rstrip("/") or "/" + await self._client.change_directory(self._root) + + async def download(self, remote_path: str) -> bytes: + assert self._client is not None + buffer = io.BytesIO() + async with self._client.download_stream(remote_path) as stream: + async for block in stream.iter_by_block(): + buffer.write(block) + return buffer.getvalue() + + async def upload(self, remote_path: str, data: bytes) -> None: + assert self._client is not None + async with self._client.upload_stream(remote_path) as stream: + await stream.write(data) + + async def remove(self, remote_path: str) -> None: + assert self._client is not None + with contextlib.suppress(Exception): + await self._client.remove_file(remote_path) + + async def list_files(self) -> list[FtpEntry]: + assert self._client is not None + entries: list[FtpEntry] = [] + async for path, info in self._client.list(recursive=False): + entries.append(FtpEntry(path=str(path), size=int(info.get("size", 0) or 0))) + return entries + + async def close(self) -> None: + if self._client is not None: + with contextlib.suppress(Exception): + await self._client.quit() + self._client = None + + +class FtpDeployer: + """``DeployTargetTester`` and the production deploy itself, over FTP/FTPS.""" + + def __init__( + self, + *, + transport_factory: TransportFactory = AsyncFtpTransport, + 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: + """Log in, ``cwd`` the target path, and write/delete a probe file.""" + try: + assert_remote_host_allowed( + f"ftp://{target.host}:{target.port}", allow_private=self.allow_private_hosts + ) + except Exception as exc: # noqa: BLE001 + 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 or "", + tls=target.kind in ("ftps",), + ) + await transport.cwd(target.remote_path) + await transport.upload(PROBE_FILENAME, b"smart-org") + await transport.remove(PROBE_FILENAME) + except Exception as exc: # noqa: BLE001 + return DeployCheck(ok=False, message=_ftp_error_message(exc)) + finally: + with contextlib.suppress(Exception): + await transport.close() + + return DeployCheck( + ok=True, + message=( + "اتصال و مسیر مقصد تأیید شد. توجه: FTP امکان تأیید کلید میزبان را " + "ندارد و تغییر فایل‌ها اتمی نیست." + ), + ) + + # -- deploy -------------------------------------------------------------- + + async def deploy( + self, + *, + target: DeployTargetSpec, + artifact_path: str, + sha: str, + organization_id: str = "", + project_id: str = "", + build_dir: str | None = None, + on_output: Callable[[str], Awaitable[None]] | None = None, + ) -> tuple[bool, str, str]: + """Back up what will be overwritten, then upload the new build. + + Returns ``(ok, backup_path, message)``. The backup path is returned even + on success: it is what a later rollback needs, and it is recorded in the + deploy row. + """ + if not os.path.isfile(artifact_path): + return False, "", "فایل بستهٔ انتشار پیدا نشد." + + # The artifact is a tar.gz of the whole release; the deployable build is + # a directory inside it. Unpack first, then choose. + extracted = _extract(artifact_path, self.artifacts_root, organization_id, project_id) + if extracted is None: + return False, "", "استخراج بستهٔ انتشار ناموفق بود." + + source_dir = _pick_build_dir(extracted, build_dir) + if source_dir is None: + return False, "", "پوشهٔ خروجی ساخت در بستهٔ انتشار پیدا نشد." + + transport = self._factory() + try: + await transport.connect( + host=target.host, + port=target.port, + username=target.username, + password=target.secret or "", + tls=target.kind == "ftps", + ) + await transport.cwd(target.remote_path) + + backup = await self._backup( + transport, + target=target, + organization_id=organization_id, + project_id=project_id, + sha=sha, + ) + if on_output is not None and backup: + await on_output(f"[deploy] پشتیبان در {backup} ذخیره شد.\n") + + uploaded = 0 + for relative, data in _walk(source_dir): + remote = relative + await transport.upload(remote, data) + uploaded += 1 + if on_output is not None: + await on_output(f"[deploy] {uploaded} فایل بارگذاری شد.\n") + + return True, backup, "استقرار در production انجام شد (FTP)." + except Exception as exc: # noqa: BLE001 + logger.exception("ftp deploy failed") + return False, "", _ftp_error_message(exc) + finally: + with contextlib.suppress(Exception): + await transport.close() + + async def _backup( + self, + transport: FtpTransport, + *, + target: DeployTargetSpec, + organization_id: str, + project_id: str, + sha: str, + ) -> str: + """Download every file we are about to overwrite, into one archive.""" + try: + entries = await transport.list_files() + except Exception: # noqa: BLE001 - an empty directory has nothing to back up + return "" + + buffer = io.BytesIO() + captured = 0 + with tarfile.open(fileobj=buffer, mode="w:gz") as tar: + for entry in entries: + name = os.path.basename(entry.path.rstrip("/")) + if not name or name == PROBE_FILENAME: + continue + try: + data = await transport.download(entry.path) + except Exception: # noqa: BLE001 - a directory entry, or unreadable + continue + info = tarfile.TarInfo(name=name) + info.size = len(data) + info.mtime = int(self._clock()) + tar.addfile(info, io.BytesIO(data)) + captured += 1 + + if not captured: + return "" + + stamp = time.strftime("%Y%m%dT%H%M%S", time.gmtime(self._clock())) + directory = os.path.join( + self.artifacts_root, organization_id or "unknown", project_id or "unknown" + ) + os.makedirs(directory, exist_ok=True) + path = os.path.join(directory, f"ftp-backup-{stamp}-{(sha or '')[:7]}.tar.gz") + with open(path, "wb") as handle: + handle.write(buffer.getvalue()) + return path + + # -- rollback ------------------------------------------------------------ + + async def rollback( + self, + *, + target: DeployTargetSpec, + backup_path: str, + on_output: Callable[[str], Awaitable[None]] | None = None, + ) -> tuple[bool, str]: + """Re-upload a backup archive over the live directory.""" + if not backup_path or not os.path.isfile(backup_path): + return False, "پشتیبان بازگردانی پیدا نشد." + + transport = self._factory() + try: + await transport.connect( + host=target.host, + port=target.port, + username=target.username, + password=target.secret or "", + tls=target.kind == "ftps", + ) + await transport.cwd(target.remote_path) + restored = 0 + with tarfile.open(backup_path, "r:gz") as tar: + for member in tar.getmembers(): + if not member.isfile(): + continue + extracted = tar.extractfile(member) + if extracted is None: + continue + await transport.upload(member.name, extracted.read()) + restored += 1 + if on_output is not None: + await on_output(f"[rollback] {restored} فایل بازگردانی شد.\n") + return True, f"بازگردانی {restored} فایل انجام شد." + except Exception as exc: # noqa: BLE001 + logger.exception("ftp rollback failed") + return False, _ftp_error_message(exc) + finally: + with contextlib.suppress(Exception): + await transport.close() + + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- + + +def _extract( + artifact_path: str, artifacts_root: str, organization_id: str, project_id: str +) -> str | None: + """Unpack a release archive into a scratch directory and return its path.""" + import shutil + + stamp = time.strftime("%Y%m%dT%H%M%S", time.gmtime()) + directory = os.path.join( + artifacts_root, organization_id or "unknown", project_id or "unknown", f"extract-{stamp}" + ) + shutil.rmtree(directory, ignore_errors=True) + os.makedirs(directory, exist_ok=True) + try: + with tarfile.open(artifact_path, "r:gz") as tar: + tar.extractall(directory, filter=_safe_member) + except (tarfile.TarError, OSError): + return None + return directory + + +def _safe_member(member: tarfile.TarInfo, dest: str) -> tarfile.TarInfo | None: + """Sanitise an archive member, or skip it. + + An uploaded artifact is not trusted. Two things are refused outright: + + * an absolute name or one containing ``..`` — the classic way a tarball + writes outside the directory it is extracted into; and + * the archive's own root entry (``.``), which would ``mkdir`` the extraction + directory itself. That is harmless on a normal filesystem but raises here, + and skipping it is correct regardless: the root already exists. + """ + name = (member.name or "").replace("\\", "/") + name = name.lstrip("/") + while name.startswith("./"): + name = name[2:] + if name in ("", "."): + return None + parts = [part for part in name.split("/") if part not in ("", ".")] + if not parts or ".." in parts: + return None + if member.issym() or member.islnk(): + # A link target is another way out of the extraction directory. + target = (member.linkname or "").lstrip("/") + if os.path.isabs(target) or ".." in target.split("/"): + return None + member.name = "/".join(parts) + return member + + +def _pick_build_dir(artifact_path: str, explicit: str | None) -> str | None: + """The directory inside the artifact that holds the deployable build. + + A build output is a *subdirectory* of the release archive (`dist/`, `build/`, + `public/`, `out/`), not the archive itself. Uploading the whole repository + over a static site is how a deploy publishes source code by accident. + """ + if not os.path.isdir(artifact_path): + return None + if explicit: + candidate = os.path.join(artifact_path, explicit) + return candidate if os.path.isdir(candidate) else None + for name in DEFAULT_BUILD_DIRS: + candidate = os.path.join(artifact_path, name) + if os.path.isdir(candidate): + return candidate + # A static site has no build step: its root *is* the output. + return artifact_path + + +def _walk(root: str): + """Yield ``(relative_path, bytes)`` for every file under ``root``.""" + for directory, _dirs, files in os.walk(root): + for name in files: + absolute = os.path.join(directory, name) + relative = os.path.relpath(absolute, root).replace(os.sep, "/") + try: + with open(absolute, "rb") as handle: + yield relative, handle.read() + except OSError: + continue + + +def _ftp_error_message(exc: BaseException) -> str: + """A plain Persian message. Never the raw exception.""" + text = str(exc).lower() + if "530" in text or "login" in text or "auth" in text: + return "ورود به سرور مقصد ناموفق بود." + if "550" in text or "not found" in text or "no such" in text: + return "مسیر مقصد روی سرور پیدا نشد." + if "timeout" in text or "timed out" in text: + return "اتصال به سرور مقصد در زمان مجاز برقرار نشد." + if "refused" in text or "unreachable" in text: + return "اتصال به سرور مقصد برقرار نشد." + if "certificate" in text or "ssl" in text or "tls" in text: + return "برقراری اتصال امن (FTPS) ناموفق بود." + return redact_text("استقرار روی سرور مقصد ناموفق بود.") + + +__all__ = [ + "CONNECT_TIMEOUT_SECONDS", + "DEFAULT_BUILD_DIRS", + "PROBE_FILENAME", + "TRANSFER_TIMEOUT_SECONDS", + "AsyncFtpTransport", + "FtpDeployer", + "FtpEntry", + "FtpTransport", +] + diff --git a/apps/api/app/infrastructure/adapters/deploy_ssh.py b/apps/api/app/infrastructure/adapters/deploy_ssh.py new file mode 100644 index 0000000..91af9f9 --- /dev/null +++ b/apps/api/app/infrastructure/adapters/deploy_ssh.py @@ -0,0 +1,475 @@ +"""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 +``/releases/-/`` 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", +] diff --git a/apps/api/app/infrastructure/adapters/deploy_target.py b/apps/api/app/infrastructure/adapters/deploy_target.py new file mode 100644 index 0000000..9f47512 --- /dev/null +++ b/apps/api/app/infrastructure/adapters/deploy_target.py @@ -0,0 +1,56 @@ +"""The deploy-target tester that dispatches on ``kind``. + +``DeployTargetTester`` is one contract with two very different implementations, +and the wizard does not care which one it is talking to: it hands over a +:class:`DeployTargetSpec` and wants a yes/no and a plain-language reason. + +The dispatcher exists so the wizard never has to know that `ssh` and `sftp` are +the same protocol while `ftp` is a different world with no host key, no atomic +switch and a mandatory backup. The *differences* are documented on the deployers +themselves; the caller sees one shape. +""" + +from __future__ import annotations + +from app.application.ports.deploy_target import ( + DEPLOY_KINDS, + DeployCheck, + DeployTargetSpec, +) +from app.core.config import settings +from app.core.errors import ValidationError +from app.infrastructure.adapters.deploy_ftp import FtpDeployer +from app.infrastructure.adapters.deploy_ssh import SshDeployer + + +class DispatchingDeployTargetTester: + """``test_connection`` routed to the right transport for ``spec.kind``.""" + + def __init__( + self, + *, + ssh: SshDeployer | None = None, + ftp: FtpDeployer | None = None, + ) -> None: + allow_private = settings.deploy_allow_private_hosts + self.ssh = ssh or SshDeployer(allow_private_hosts=allow_private) + self.ftp = ftp or FtpDeployer(allow_private_hosts=allow_private) + + async def test_connection(self, *, target: DeployTargetSpec) -> DeployCheck: + if target.kind not in DEPLOY_KINDS: + raise ValidationError( + "نوع مقصد استقرار پشتیبانی نمی‌شود؛ مقادیر مجاز: " + + "، ".join(DEPLOY_KINDS) + ) + if not target.remote_path.strip(): + raise ValidationError("مسیر مقصد روی سرور مشخص نشده است.") + if target.kind in ("ssh", "sftp"): + return await self.ssh.test_connection(target=target) + return await self.ftp.test_connection(target=target) + + def deployer_for(self, kind: str): + """The object that can actually deploy to this kind of target.""" + return self.ssh if kind in ("ssh", "sftp") else self.ftp + + +__all__ = ["DispatchingDeployTargetTester"] diff --git a/apps/api/app/infrastructure/db/models_delivery.py b/apps/api/app/infrastructure/db/models_delivery.py index 23a4941..e2a3ccd 100644 --- a/apps/api/app/infrastructure/db/models_delivery.py +++ b/apps/api/app/infrastructure/db/models_delivery.py @@ -121,8 +121,46 @@ class AppRuntimeRow(Base): ) +class ProductionDeployRow(Base): + """One human-triggered production deploy (and its rollbacks). + + ``log_tail`` is bounded exactly like a pipeline step's: the full output is + streamed, the tail is kept for the post-mortem. ``rolled_back_from`` makes a + rollback its own row rather than an edit of the deploy it undid — "what was + live when?" must be answerable from the table alone. + """ + + __tablename__ = "production_deploys" + __table_args__ = (Index("ix_prod_deploy_project_created", "project_id", "created_at"),) + + id: Mapped[str] = pk() + organization_id: Mapped[str] = mapped_column( + ForeignKey("organizations.id", ondelete="CASCADE"), nullable=False, index=True + ) + project_id: Mapped[str] = mapped_column( + ForeignKey("projects.id", ondelete="CASCADE"), nullable=False, index=True + ) + release_id: Mapped[str | None] = mapped_column( + ForeignKey("releases.id", ondelete="SET NULL"), nullable=True + ) + target_id: Mapped[str | None] = mapped_column( + ForeignKey("deploy_targets.id", ondelete="SET NULL"), nullable=True + ) + kind: Mapped[str] = mapped_column(String(8), nullable=False) + host: Mapped[str] = mapped_column(String(255), default="", nullable=False) + status: Mapped[str] = mapped_column(String(16), default="queued", nullable=False) + message: Mapped[str] = mapped_column(Text, default="", nullable=False) + log_tail: Mapped[str] = mapped_column(Text, default="", nullable=False) + rolled_back_from: Mapped[str | None] = mapped_column(String(32), nullable=True) + backup_path: Mapped[str] = mapped_column(String(512), default="", nullable=False) + started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + created_at: Mapped[datetime] = created() + + __all__ = [ "AppRuntimeRow", + "ProductionDeployRow", "MAX_LOG_TAIL_CHARS", "PipelineRunRow", "PipelineStepRow", diff --git a/apps/api/app/infrastructure/repositories/delivery.py b/apps/api/app/infrastructure/repositories/delivery.py index 62a90eb..4ee6df3 100644 --- a/apps/api/app/infrastructure/repositories/delivery.py +++ b/apps/api/app/infrastructure/repositories/delivery.py @@ -8,8 +8,10 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.domain.base import utcnow from app.domain.delivery.models import ( AppRuntime, + DeployStatus, PipelineRun, PipelineStep, + ProductionDeploy, Release, RunStatus, StepStatus, @@ -18,6 +20,7 @@ from app.infrastructure.db.models_delivery import ( AppRuntimeRow, PipelineRunRow, PipelineStepRow, + ProductionDeployRow, ReleaseRow, ) @@ -440,11 +443,131 @@ class AppRuntimeRepository: await self.session.flush() +def _deploy(row: ProductionDeployRow) -> ProductionDeploy: + return ProductionDeploy( + id=row.id, + organization_id=row.organization_id, + project_id=row.project_id, + kind=row.kind, + host=row.host, + release_id=row.release_id, + target_id=row.target_id, + status=row.status, + message=row.message, + log_tail=row.log_tail, + rolled_back_from=row.rolled_back_from, + backup_path=row.backup_path, + started_at=row.started_at, + finished_at=row.finished_at, + created_at=row.created_at, + ) + + +class ProductionDeployRepository: + """The production deploy ledger.""" + + def __init__(self, session: AsyncSession) -> None: + self.session = session + + async def get(self, deploy_id: str) -> ProductionDeploy | None: + row = await self.session.get(ProductionDeployRow, deploy_id) + return _deploy(row) if row else None + + async def get_for_org(self, organization_id: str, deploy_id: str) -> ProductionDeploy | None: + row = await self.session.scalar( + select(ProductionDeployRow).where( + ProductionDeployRow.id == deploy_id, + ProductionDeployRow.organization_id == organization_id, + ) + ) + return _deploy(row) if row else None + + async def list_for_project( + self, *, project_id: str, limit: int = 20 + ) -> list[ProductionDeploy]: + stmt = ( + select(ProductionDeployRow) + .where(ProductionDeployRow.project_id == project_id) + .order_by(ProductionDeployRow.created_at.desc()) + .limit(min(max(limit, 1), 100)) + ) + return [_deploy(r) for r in (await self.session.execute(stmt)).scalars()] + + async def active_for_project(self, project_id: str) -> ProductionDeploy | None: + """One production deploy per project at a time. + + Two concurrent deploys to the same target would interleave their writes + on a shared host, which is exactly the failure FTP has no way to undo. + """ + row = await self.session.scalar( + select(ProductionDeployRow) + .where( + ProductionDeployRow.project_id == project_id, + ProductionDeployRow.status.in_( + [DeployStatus.QUEUED.value, DeployStatus.RUNNING.value] + ), + ) + .limit(1) + ) + return _deploy(row) if row else None + + async def create(self, deploy: ProductionDeploy) -> ProductionDeploy: + row = ProductionDeployRow( + id=deploy.id, + organization_id=deploy.organization_id, + project_id=deploy.project_id, + release_id=deploy.release_id, + target_id=deploy.target_id, + kind=deploy.kind, + host=deploy.host, + status=deploy.status, + message=deploy.message, + log_tail=deploy.log_tail, + rolled_back_from=deploy.rolled_back_from, + backup_path=deploy.backup_path, + started_at=deploy.started_at, + finished_at=deploy.finished_at, + created_at=deploy.created_at, + ) + self.session.add(row) + await self.session.flush() + return _deploy(row) + + async def finish( + self, + deploy_id: str, + *, + status: str, + message: str = "", + log_tail: str = "", + backup_path: str = "", + ) -> None: + row = await self.session.get(ProductionDeployRow, deploy_id) + if row is None: + return + row.status = status + row.message = message[:2_000] + row.log_tail = log_tail[-65_536:] + if backup_path: + row.backup_path = backup_path + row.finished_at = utcnow() + await self.session.flush() + + async def mark_running(self, deploy_id: str) -> None: + row = await self.session.get(ProductionDeployRow, deploy_id) + if row is None: + return + row.status = DeployStatus.RUNNING.value + row.started_at = utcnow() + await self.session.flush() + + __all__ = [ "ACTIVE_STATUSES", "AppRuntimeRepository", "PipelineRunRepository", "PipelineStepRepository", + "ProductionDeployRepository", "ReleaseRepository", "StepStatus", ] diff --git a/apps/api/app/infrastructure/repositories/projects.py b/apps/api/app/infrastructure/repositories/projects.py index 5c2f718..551cd36 100644 --- a/apps/api/app/infrastructure/repositories/projects.py +++ b/apps/api/app/infrastructure/repositories/projects.py @@ -14,6 +14,7 @@ from app.domain.projects.models import ( ProjectGitRepo, ) from app.infrastructure.db.models import ( + DeployTargetRow, PreviewDomainRow, PreviewPortRow, ProjectExternalGitRow, @@ -164,6 +165,61 @@ class ProjectGitCredentialRepository: return credential +class DeployTargetRepository: + """Persistence for a project's production deploy target. + + The secret is stored encrypted and is only ever handed back through + :meth:`get_secret`, so a code path that merely *renders* a target never + touches the plaintext. + """ + + def __init__(self, session: AsyncSession) -> None: + self.session = session + + async def get_for_project(self, project_id: str) -> DeployTargetRow | None: + return await self.session.scalar( + select(DeployTargetRow) + .where(DeployTargetRow.project_id == project_id, DeployTargetRow.is_active.is_(True)) + .order_by(DeployTargetRow.created_at.desc()) + .limit(1) + ) + + async def list_for_project(self, project_id: str) -> list[DeployTargetRow]: + stmt = ( + select(DeployTargetRow) + .where(DeployTargetRow.project_id == project_id) + .order_by(DeployTargetRow.created_at.desc()) + ) + return list((await self.session.execute(stmt)).scalars()) + + async def upsert(self, *, values: dict) -> DeployTargetRow: + """Create or replace the single active target of a project.""" + project_id = str(values["project_id"]) + existing = await self.get_for_project(project_id) + if existing is None: + row = DeployTargetRow(**values) + self.session.add(row) + else: + row = existing + for key, value in values.items(): + setattr(row, key, value) + await self.session.flush() + return row + + async def record_test( + self, *, target_id: str, ok: bool, message: str, fingerprint: str | None = None + ) -> None: + row = await self.session.get(DeployTargetRow, target_id) + if row is None: + return + row.last_tested_at = utcnow() + row.last_test_ok = ok + row.last_test_message = message[:500] + if fingerprint: + row.host_fingerprint = fingerprint + await self.session.flush() + + class ProjectRepository: def __init__(self, session: AsyncSession) -> None: self.session = session diff --git a/apps/api/app/interfaces/http/deps.py b/apps/api/app/interfaces/http/deps.py index 165540b..22acf77 100644 --- a/apps/api/app/interfaces/http/deps.py +++ b/apps/api/app/interfaces/http/deps.py @@ -204,3 +204,12 @@ async def get_delivery_service( from app.application.container import build_delivery_service return build_delivery_service(session) + + +async def get_deploy_service( + session: AsyncSession = Depends(get_session), +): + """Build the production deploy service with the configured transports.""" + from app.application.container import build_deploy_service + + return build_deploy_service(session) diff --git a/apps/api/app/interfaces/http/routers/delivery.py b/apps/api/app/interfaces/http/routers/delivery.py index 3b5b4f8..ee488f2 100644 --- a/apps/api/app/interfaces/http/routers/delivery.py +++ b/apps/api/app/interfaces/http/routers/delivery.py @@ -21,12 +21,14 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.application.services.audit_service import record_audit from app.application.services.delivery_service import DeliveryService +from app.application.services.deploy_service import DeployService from app.core.database import get_session from app.domain.identity.models import CAN_MANAGE_PROJECTS from app.interfaces.http.deps import ( OrgContext, client_meta, get_delivery_service, + get_deploy_service, get_org_context, ) @@ -248,3 +250,210 @@ async def rollback_demo( ip=ip, user_agent=ua, ) return _release_out(release) + + +# --------------------------------------------------------------------------- +# Production (always human-triggered, always confirmed) +# --------------------------------------------------------------------------- + + +class ProductionDeployOut(BaseModel): + id: str + project_id: str + kind: str + host: str + status: str + message: str + log_tail: str + release_id: str | None + rolled_back_from: str | None + started_at: datetime | None + finished_at: datetime | None + created_at: datetime + + +class ProductionTargetOut(BaseModel): + """The target as a *user* sees it: never a secret, only a hint that one exists.""" + + kind: str + host: str + port: int + username: str + auth_type: str + remote_path: str + has_secret: bool + host_fingerprint: str | None + last_tested_at: datetime | None + last_test_ok: bool | None + last_test_message: str | None + + +class ProductionOverviewOut(BaseModel): + target: ProductionTargetOut | None + deploys: list[ProductionDeployOut] + + +class DeployConfirmRequest(BaseModel): + project_id: str = Field(min_length=1, max_length=32) + #: The organization's slug, typed back. Not ceremony: this is the only + #: action in the product the platform cannot undo. + confirm: str = Field(min_length=1, max_length=120) + release_id: str | None = Field(default=None, max_length=32) + + +class RollbackConfirmRequest(BaseModel): + project_id: str = Field(min_length=1, max_length=32) + confirm: str = Field(min_length=1, max_length=120) + deploy_id: str | None = Field(default=None, max_length=32) + + +class TargetTestOut(BaseModel): + ok: bool + message: str + + +def _deploy_out(deploy) -> ProductionDeployOut: # noqa: ANN001 - a domain dataclass + return ProductionDeployOut( + id=deploy.id, project_id=deploy.project_id, kind=deploy.kind, host=deploy.host, + status=deploy.status, message=deploy.message, log_tail=deploy.log_tail, + release_id=deploy.release_id, rolled_back_from=deploy.rolled_back_from, + started_at=deploy.started_at, finished_at=deploy.finished_at, + created_at=deploy.created_at, + ) + + +def _target_out(target) -> ProductionTargetOut: # noqa: ANN001 + return ProductionTargetOut( + kind=target.kind, host=target.host, port=target.port, username=target.username, + auth_type=target.auth_type, remote_path=target.remote_path, + # The secret itself is never returned — only that one is configured. + has_secret=bool(target.secret_encrypted), + host_fingerprint=target.host_fingerprint, + last_tested_at=target.last_tested_at, last_test_ok=target.last_test_ok, + last_test_message=target.last_test_message, + ) + + +@router.get("/production", response_model=ProductionOverviewOut) +async def production_overview( + project_id: str = Query(min_length=1), + context: OrgContext = Depends(get_org_context), + deploys: DeployService = Depends(get_deploy_service), +) -> ProductionOverviewOut: + """The target summary and the deploy history (admin+ — it names a host).""" + context.require(CAN_MANAGE_PROJECTS) + target = await deploys.get_target( + organization_id=context.org_id, project_id=project_id + ) + history = await deploys.list_deploys( + organization_id=context.org_id, project_id=project_id + ) + return ProductionOverviewOut( + target=_target_out(target) if target else None, + deploys=[_deploy_out(d) for d in history], + ) + + +@router.post("/production/test", response_model=TargetTestOut) +async def test_production_target( + payload: ManualRunRequest, + context: OrgContext = Depends(get_org_context), + deploys: DeployService = Depends(get_deploy_service), +) -> TargetTestOut: + """Test the configured target and record the result (admin+).""" + context.require(CAN_MANAGE_PROJECTS) + ok, message = await deploys.test_target( + organization_id=context.org_id, + project_id=payload.project_id, + actor_user_id=context.user.id, + ) + return TargetTestOut(ok=ok, message=message) + + +@router.post("/production/deploy", response_model=ProductionDeployOut) +async def deploy_to_production( + payload: DeployConfirmRequest, + request: Request, + context: OrgContext = Depends(get_org_context), + deploys: DeployService = Depends(get_deploy_service), + session: AsyncSession = Depends(get_session), +) -> ProductionDeployOut: + """Ship a release to the customer's server (org owner/admin, confirmed).""" + context.require(CAN_MANAGE_PROJECTS) + deploy = await deploys.deploy_production( + organization_id=context.org_id, + project_id=payload.project_id, + actor_user_id=context.user.id, + confirm=payload.confirm, + release_id=payload.release_id, + ) + ip, ua = client_meta(request) + await record_audit( + session, action="deploy.production", organization_id=context.org_id, + actor_user_id=context.user.id, target_type="production_deploy", target_id=deploy.id, + metadata={ + "project_id": deploy.project_id, + "kind": deploy.kind, + "status": deploy.status, + "release_id": deploy.release_id, + }, + ip=ip, user_agent=ua, + ) + await _emit_deploy_status(context.org_id, deploy) + return _deploy_out(deploy) + + +@router.post("/production/rollback", response_model=ProductionDeployOut) +async def rollback_production( + payload: RollbackConfirmRequest, + request: Request, + context: OrgContext = Depends(get_org_context), + deploys: DeployService = Depends(get_deploy_service), + session: AsyncSession = Depends(get_session), +) -> ProductionDeployOut: + """Undo a production deploy (org owner/admin, confirmed, audited).""" + context.require(CAN_MANAGE_PROJECTS) + deploy = await deploys.rollback_production( + organization_id=context.org_id, + project_id=payload.project_id, + actor_user_id=context.user.id, + confirm=payload.confirm, + deploy_id=payload.deploy_id, + ) + ip, ua = client_meta(request) + await record_audit( + session, action="deploy.production_rollback", organization_id=context.org_id, + actor_user_id=context.user.id, target_type="production_deploy", target_id=deploy.id, + metadata={ + "project_id": deploy.project_id, + "kind": deploy.kind, + "status": deploy.status, + "rolled_back_from": deploy.rolled_back_from, + }, + ip=ip, user_agent=ua, + ) + await _emit_deploy_status(context.org_id, deploy) + return _deploy_out(deploy) + + +async def _emit_deploy_status(organization_id: str, deploy) -> None: # noqa: ANN001 + """Push the outcome on the tenant channel. Never raises.""" + import contextlib + + from app.application.ports.realtime import RealtimeEvent, get_publisher, org_channel + + with contextlib.suppress(Exception): + await get_publisher().publish( + RealtimeEvent( + type="deploy.status", + channels=(org_channel(organization_id),), + data={ + "project_id": deploy.project_id, + "deploy_id": deploy.id, + "kind": deploy.kind, + "status": deploy.status, + "message": deploy.message, + }, + organization_id=organization_id, + ) + ) diff --git a/apps/api/tests/test_deploy.py b/apps/api/tests/test_deploy.py new file mode 100644 index 0000000..fe57e7a --- /dev/null +++ b/apps/api/tests/test_deploy.py @@ -0,0 +1,787 @@ +"""§v2-A3.5 — production deploy over SSH / SFTP / FTP / FTPS (D8). + +Production is the one action the platform cannot undo, so the tests are about the +guards around it as much as the deploy itself: + +* a member who is not an org admin cannot deploy; +* the org slug has to be typed back; +* a changed host key is a hard failure; +* an SSRF target is refused before a connection is attempted; +* a secret never reaches a log, an event, a response or a database column; +* and the mechanics — release directories, the ``current`` symlink, keeping five + releases, rollback — behave the way the design says they do. + +The SSH and FTP transports are faked, so the suite needs no server. The deploy +*logic* is real. +""" + +from __future__ import annotations + +import os +import tarfile + +import pytest +import pytest_asyncio +from httpx import AsyncClient + +from app.application.ports.deploy_target import DeployCheck, DeployTargetSpec +from app.application.services.deploy_service import DeployService +from app.core.crypto import encrypt +from app.domain.delivery.models import DeployStatus +from app.domain.identity.models import PlatformRole +from app.infrastructure.adapters.deploy_ftp import FtpDeployer, FtpEntry +from app.infrastructure.adapters.deploy_ssh import ( + CURRENT_LINK, + KEEP_RELEASES, + CommandOutcome, + SshDeployer, +) +from app.infrastructure.repositories.delivery import ( + ProductionDeployRepository, + ReleaseRepository, +) +from app.infrastructure.repositories.identity import OrganizationRepository +from app.infrastructure.repositories.projects import ( + DeployTargetRepository, + ProjectRepository, +) +from tests.conftest import auth_header, signup + +SECRET = "ssh-password-that-must-never-leak" + + +# --------------------------------------------------------------------------- +# fake transports +# --------------------------------------------------------------------------- + + +class FakeSshTransport: + """An in-memory SSH server: a filesystem plus a scripted command table.""" + + def __init__(self, *, fingerprint: str = "SHA256:abc123", fail_on: str = "") -> None: + self.commands: list[str] = [] + self.uploads: list[tuple[str, str]] = [] + self.fingerprint = fingerprint + self.fail_on = fail_on + self.closed = False + #: The remote filesystem, as a set of paths. + self.files: set[str] = set() + + async def connect(self, **kwargs) -> None: # noqa: ANN003 + self.connect_kwargs = kwargs + if self.fail_on == "connect": + raise OSError("permission denied") + + def host_key_fingerprint(self) -> str: + return self.fingerprint + + async def run(self, command: str, *, timeout: float = 600.0) -> CommandOutcome: + self.commands.append(command) + if self.fail_on and self.fail_on in command: + return CommandOutcome(exit_code=1, output="failed on purpose") + + if "ls -1t" in command and "releases" in command: + releases = sorted( + {p.split("/releases/")[1].split("/")[0] for p in self.files if "/releases/" in p}, + reverse=True, + ) + return CommandOutcome(exit_code=0, output="\n".join(releases)) + if "test -w" in command: + return CommandOutcome(exit_code=0, output="") + if "ln -sfn" in command: + # Record the switch so a test can assert the pointer moved. + for part in command.split(): + if part.startswith("releases/"): + self.files.add(f"/srv/app/{CURRENT_LINK}->{part}") + return CommandOutcome(exit_code=0, output="") + if "tail -n" in command: + self.files = { + p for p in self.files if not p.endswith(tuple( + f"release-{i}" for i in range(1, 20) + )) + } + return CommandOutcome(exit_code=0, output="") + return CommandOutcome(exit_code=0, output="") + + async def upload(self, *, local_path: str, remote_path: str) -> None: + self.uploads.append((local_path, remote_path)) + self.files.add(remote_path) + + async def close(self) -> None: + self.closed = True + + +class FakeFtpTransport: + def __init__(self, *, fail_on: str = "") -> None: + self.commands: list[str] = [] + self.uploaded: dict[str, bytes] = {} + self.removed: list[str] = [] + self.fail_on = fail_on + self.cwd_path = "" + self.connected = False + + async def connect(self, **kwargs) -> None: # noqa: ANN003 + self.connect_kwargs = kwargs + if self.fail_on == "connect": + raise OSError("530 login incorrect") + self.connected = True + + async def cwd(self, path: str) -> None: + if self.fail_on == "cwd": + raise OSError("550 no such directory") + self.cwd_path = path + + async def download(self, remote_path: str) -> bytes: + return self.uploaded.get(remote_path, b"old-content") + + async def upload(self, remote_path: str, data: bytes) -> None: + if self.fail_on == "upload": + raise OSError("552 quota exceeded") + self.uploaded[remote_path] = data + + async def remove(self, remote_path: str) -> None: + self.removed.append(remote_path) + self.uploaded.pop(remote_path, None) + + async def list_files(self) -> list[FtpEntry]: + return [FtpEntry(path=name, size=len(data)) for name, data in self.uploaded.items()] + + async def close(self) -> None: + self.connected = False + + +class FakeDeployTargetTester: + """Dispatches to injected deployers, like the real one but with no network.""" + + def __init__(self, *, ssh: SshDeployer, ftp: FtpDeployer) -> None: + self.ssh = ssh + self.ftp = ftp + self.calls: list[DeployTargetSpec] = [] + + async def test_connection(self, *, target: DeployTargetSpec) -> DeployCheck: + self.calls.append(target) + if target.kind in ("ssh", "sftp"): + return await self.ssh.test_connection(target=target) + return await self.ftp.test_connection(target=target) + + def deployer_for(self, kind: str): + return self.ssh if kind in ("ssh", "sftp") else self.ftp + + +# --------------------------------------------------------------------------- +# fixtures +# --------------------------------------------------------------------------- + + +@pytest_asyncio.fixture(autouse=True) +async def _platform_seed(seeded): + return seeded + + +def _artifact(tmp_path, *, build_dir: str = "") -> str: + """A release archive shaped like the pipeline produces.""" + source = tmp_path / "artifact-src" + target = source / build_dir if build_dir else source + target.mkdir(parents=True, exist_ok=True) + (target / "index.html").write_text("

hello

") + (target / "app.js").write_text("console.log('hi')") + if build_dir: + (source / "README.md").write_text("not deployed") + + archive = tmp_path / "release.tar.gz" + with tarfile.open(archive, "w:gz") as tar: + # Add the children rather than the directory itself: a bare "." member + # is the archive root, which some extractors try to mkdir. + for child in sorted(source.iterdir()): + tar.add(child, arcname=child.name) + return str(archive) + + +@pytest_asyncio.fixture +async def tenant(client: AsyncClient, session_factory, tmp_path): + """An org + app + live release + a configured SSH deploy target.""" + token = ( + await signup( + client, + "deploy-owner@example.com", + name="مدیر استقرار", + platform_role=PlatformRole.OWNER, + ) + )["access_token"] + headers = auth_header(token) + org = ( + await client.post("/api/organizations", json={"name": "شرکت استقرار"}, headers=headers) + ).json() + org_id, slug = org["id"], org["slug"] + project_id = ( + await client.post( + f"/api/organizations/{org_id}/projects", + json={"name": "اپ استقرار"}, + headers=headers, + ) + ).json()["id"] + + artifact = _artifact(tmp_path, build_dir="dist") + + async with session_factory() as session: + from app.domain.delivery.models import PipelineRun, Release + from app.infrastructure.repositories.delivery import PipelineRunRepository + + # A release points at the pipeline run that produced it, so the run has + # to exist: the foreign key is real, not decorative. + run = await PipelineRunRepository(session).queue( + PipelineRun( + organization_id=org_id, + project_id=project_id, + sha="abc1234def", + status="succeeded", + ) + ) + release = await ReleaseRepository(session).create( + Release( + organization_id=org_id, + project_id=project_id, + run_id=run.id, + sha="abc1234def", + artifact_path=artifact, + is_demo_live=True, + ) + ) + await DeployTargetRepository(session).upsert( + values={ + "organization_id": org_id, + "project_id": project_id, + "kind": "ssh", + "host": "deploy.example.com", + "port": 22, + "username": "deployer", + "auth_type": "password", + "secret_encrypted": encrypt(SECRET), + "remote_path": "/srv/app", + "post_deploy_command": "systemctl reload app", + "is_active": True, + } + ) + await session.commit() + + return { + "org_id": org_id, + "slug": slug, + "project_id": project_id, + "headers": headers, + "release_id": release.id, + "artifact": artifact, + } + + +def _service(session, *, ssh: FakeSshTransport, ftp: FakeFtpTransport | None = None) -> tuple[ + DeployService, FakeDeployTargetTester +]: + ssh_deployer = SshDeployer( + transport_factory=lambda: ssh, artifacts_root="/tmp/artifacts", allow_private_hosts=False + ) + ftp_deployer = FtpDeployer( + transport_factory=lambda: ftp or FakeFtpTransport(), + artifacts_root="/tmp/artifacts", + allow_private_hosts=False, + ) + tester = FakeDeployTargetTester(ssh=ssh_deployer, ftp=ftp_deployer) + service = DeployService( + projects=ProjectRepository(session), + organizations=OrganizationRepository(session), + targets=DeployTargetRepository(session), + deploys=ProductionDeployRepository(session), + releases=ReleaseRepository(session), + runtimes=None, # not needed for the deploy paths under test + tester=tester, + allow_private_hosts=False, + ) + return service, tester + + +# --------------------------------------------------------------------------- +# the SSH deploy mechanics +# --------------------------------------------------------------------------- + + +async def test_ssh_deploy_uploads_extracts_and_switches(tmp_path): + transport = FakeSshTransport() + deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True) + + ok, release_dir, message = await deployer.deploy( + target=DeployTargetSpec( + kind="ssh", host="deploy.example.com", port=22, username="deployer", + secret="pw", remote_path="/srv/app", + ), + artifact_path=_artifact(tmp_path, build_dir="dist"), + sha="abc1234def", + post_deploy_command="systemctl reload app", + ) + + assert ok is True, message + assert release_dir.startswith("/srv/app/releases/") + assert transport.uploads, "the artifact must be uploaded" + joined = "\n".join(transport.commands) + assert "tar -xzf" in joined, "the artifact must be extracted on the target" + assert "systemctl reload app" in joined, "the post-deploy command must run" + assert "ln -sfn" in joined and CURRENT_LINK in joined, "the symlink must be switched" + assert "tail -n" in joined, "old releases must be pruned" + + +async def test_ssh_deploy_prunes_to_five_releases(tmp_path): + transport = FakeSshTransport() + deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True) + await deployer.deploy( + target=DeployTargetSpec( + kind="ssh", host="h", port=22, username="u", secret="pw", remote_path="/srv/app" + ), + artifact_path=_artifact(tmp_path), + sha="abc1234", + ) + prune = next(c for c in transport.commands if "tail -n" in c) + assert f"tail -n +{KEEP_RELEASES + 1}" in prune, ( + "keep the newest five and delete the rest" + ) + + +async def test_a_failed_post_deploy_command_does_not_switch_the_symlink(tmp_path): + """The previous release must keep serving after a failed deploy.""" + transport = FakeSshTransport(fail_on="systemctl") + deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True) + + ok, _release, message = await deployer.deploy( + target=DeployTargetSpec( + kind="ssh", host="h", port=22, username="u", secret="pw", remote_path="/srv/app" + ), + artifact_path=_artifact(tmp_path), + sha="abc1234", + post_deploy_command="systemctl reload app", + ) + + assert ok is False + assert "پس از استقرار" in message + assert not any("ln -sfn" in c for c in transport.commands), ( + "the switch must not happen when the hook failed" + ) + + +async def test_ssh_rollback_points_current_at_the_previous_release(tmp_path): + transport = FakeSshTransport() + deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True) + # Two releases exist on the target, newest first. + transport.files = { + "/srv/app/releases/20260101T000000-aaaaaaa/index.html", + "/srv/app/releases/20260102T000000-bbbbbbb/index.html", + } + + ok, target_name, message = await deployer.rollback( + target=DeployTargetSpec( + kind="ssh", host="h", port=22, username="u", secret="pw", remote_path="/srv/app" + ), + post_deploy_command="systemctl reload app", + ) + + assert ok is True, message + assert target_name == "20260101T000000-aaaaaaa" + assert any("ln -sfn" in c for c in transport.commands) + + +async def test_ssh_rollback_refuses_when_there_is_only_one_release(tmp_path): + transport = FakeSshTransport() + transport.files = {"/srv/app/releases/only-one/index.html"} + deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True) + + ok, _target, message = await deployer.rollback( + target=DeployTargetSpec( + kind="ssh", host="h", port=22, username="u", secret="pw", remote_path="/srv/app" + ) + ) + assert ok is False + assert "قبلی" in message + + +# --------------------------------------------------------------------------- +# host key pinning (TOFU) +# --------------------------------------------------------------------------- + + +async def test_a_changed_host_key_is_a_hard_failure(): + transport = FakeSshTransport(fingerprint="SHA256:CHANGED") + deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True) + + check = await deployer.test_connection( + target=DeployTargetSpec( + kind="ssh", host="h", port=22, username="u", secret="pw", + remote_path="/srv/app", known_fingerprint="SHA256:ORIGINAL", + ) + ) + + assert check.ok is False + assert "کلید میزبان" in check.message + assert check.fingerprint == "SHA256:CHANGED" + + +async def test_the_fingerprint_is_returned_for_pinning_on_first_contact(): + transport = FakeSshTransport(fingerprint="SHA256:FIRST") + deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True) + + check = await deployer.test_connection( + target=DeployTargetSpec( + kind="ssh", host="h", port=22, username="u", secret="pw", remote_path="/srv/app" + ) + ) + assert check.ok is True + assert check.fingerprint == "SHA256:FIRST" + + +async def test_a_matching_pinned_key_is_accepted(): + transport = FakeSshTransport(fingerprint="SHA256:SAME") + deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=True) + check = await deployer.test_connection( + target=DeployTargetSpec( + kind="ssh", host="h", port=22, username="u", secret="pw", + remote_path="/srv/app", known_fingerprint="SHA256:SAME", + ) + ) + assert check.ok is True + + +# --------------------------------------------------------------------------- +# SSRF +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "host", + ["127.0.0.1", "169.254.169.254", "10.0.0.5", "localhost"], +) +async def test_an_internal_target_is_refused_before_connecting(host: str): + transport = FakeSshTransport() + deployer = SshDeployer(transport_factory=lambda: transport, allow_private_hosts=False) + + check = await deployer.test_connection( + target=DeployTargetSpec( + kind="ssh", host=host, port=22, username="u", secret="pw", remote_path="/srv/app" + ) + ) + + assert check.ok is False + assert transport.commands == [], "no connection may be attempted" + assert not transport.uploads + + +async def test_an_internal_ftp_target_is_refused(): + transport = FakeFtpTransport() + deployer = FtpDeployer(transport_factory=lambda: transport, allow_private_hosts=False) + check = await deployer.test_connection( + target=DeployTargetSpec( + kind="ftp", host="127.0.0.1", port=21, username="u", secret="pw", + remote_path="/var/www", + ) + ) + assert check.ok is False + assert transport.connected is False + + +# --------------------------------------------------------------------------- +# FTP +# --------------------------------------------------------------------------- + + +async def test_ftp_deploy_backs_up_before_overwriting(tmp_path): + transport = FakeFtpTransport() + transport.uploaded = {"index.html": b"OLD", "app.js": b"OLD"} + deployer = FtpDeployer( + transport_factory=lambda: transport, artifacts_root=str(tmp_path / "artifacts") + ) + + ok, backup, message = await deployer.deploy( + target=DeployTargetSpec( + kind="ftp", host="ftp.example.com", port=21, username="u", secret="pw", + remote_path="/var/www", + ), + artifact_path=_artifact(tmp_path, build_dir="dist"), + sha="abc1234", + organization_id="org-1", + project_id="proj-1", + ) + + assert ok is True, message + assert backup and os.path.isfile(backup), "a backup must exist before anything is overwritten" + with tarfile.open(backup, "r:gz") as tar: + names = set(tar.getnames()) + assert {"index.html", "app.js"} <= names, "the backup must contain what it overwrites" + + +async def test_ftp_deploy_uploads_only_the_build_output(tmp_path): + """A static deploy must not publish the repository, only the build.""" + transport = FakeFtpTransport() + deployer = FtpDeployer( + transport_factory=lambda: transport, artifacts_root=str(tmp_path / "artifacts") + ) + await deployer.deploy( + target=DeployTargetSpec( + kind="ftp", host="h", port=21, username="u", secret="pw", remote_path="/var/www" + ), + artifact_path=_artifact(tmp_path, build_dir="dist"), + sha="abc1234", + ) + + assert "index.html" in transport.uploaded + assert "README.md" not in transport.uploaded, "the source tree must not be published" + + +async def test_ftp_rollback_re_uploads_the_backup(tmp_path): + transport = FakeFtpTransport() + transport.uploaded = {"index.html": b"OLD"} + deployer = FtpDeployer( + transport_factory=lambda: transport, artifacts_root=str(tmp_path / "artifacts") + ) + _ok, backup, _message = await deployer.deploy( + target=DeployTargetSpec( + kind="ftp", host="h", port=21, username="u", secret="pw", remote_path="/var/www" + ), + artifact_path=_artifact(tmp_path, build_dir="dist"), + sha="abc1234", + organization_id="org-1", + project_id="proj-1", + ) + + transport.uploaded = {"index.html": b"BROKEN"} + ok, message = await deployer.rollback( + target=DeployTargetSpec( + kind="ftp", host="h", port=21, username="u", secret="pw", remote_path="/var/www" + ), + backup_path=backup, + ) + + assert ok is True, message + assert transport.uploaded["index.html"] == b"OLD" + + +async def test_ftp_rollback_without_a_backup_fails_cleanly(): + transport = FakeFtpTransport() + deployer = FtpDeployer(transport_factory=lambda: transport) + ok, message = await deployer.rollback( + target=DeployTargetSpec( + kind="ftp", host="h", port=21, username="u", secret="pw", remote_path="/var/www" + ), + backup_path="", + ) + assert ok is False + assert message + + +# --------------------------------------------------------------------------- +# the service guards +# --------------------------------------------------------------------------- + + +async def test_the_confirm_slug_must_match(tenant, session_factory): + transport = FakeSshTransport() + async with session_factory() as session: + service, _tester = _service(session, ssh=transport) + with pytest.raises(Exception) as excinfo: + await service.deploy_production( + organization_id=tenant["org_id"], + project_id=tenant["project_id"], + actor_user_id="u1", + confirm="not-the-slug", + ) + assert "تأییدیه" in str(excinfo.value) + assert transport.commands == [], "nothing may be attempted on a bad confirmation" + + +async def test_a_correct_confirmation_deploys_and_records_the_row(tenant, session_factory): + transport = FakeSshTransport() + async with session_factory() as session: + service, _tester = _service(session, ssh=transport) + deploy = await service.deploy_production( + organization_id=tenant["org_id"], + project_id=tenant["project_id"], + actor_user_id="u1", + confirm=tenant["slug"], + ) + await session.commit() + + assert deploy.status == DeployStatus.SUCCEEDED.value, deploy.message + assert deploy.kind == "ssh" + async with session_factory() as session: + rows = await ProductionDeployRepository(session).list_for_project( + project_id=tenant["project_id"] + ) + assert len(rows) == 1 + assert rows[0].release_id == tenant["release_id"] + + +async def test_a_second_deploy_is_refused_while_one_is_running(tenant, session_factory): + from app.domain.delivery.models import ProductionDeploy + + transport = FakeSshTransport() + async with session_factory() as session: + repo = ProductionDeployRepository(session) + await repo.create( + ProductionDeploy( + organization_id=tenant["org_id"], + project_id=tenant["project_id"], + kind="ssh", + host="h", + status=DeployStatus.RUNNING.value, + ) + ) + await session.commit() + + service, _tester = _service(session, ssh=transport) + with pytest.raises(Exception) as excinfo: + await service.deploy_production( + organization_id=tenant["org_id"], + project_id=tenant["project_id"], + actor_user_id="u1", + confirm=tenant["slug"], + ) + assert "در حال انجام" in str(excinfo.value) + + +async def test_a_secret_never_reaches_the_log_or_the_row(tenant, session_factory): + """The one thing that must never happen: a credential in a stored field.""" + transport = FakeSshTransport() + async with session_factory() as session: + service, _tester = _service(session, ssh=transport) + deploy = await service.deploy_production( + organization_id=tenant["org_id"], + project_id=tenant["project_id"], + actor_user_id="u1", + confirm=tenant["slug"], + ) + await session.commit() + + assert SECRET not in deploy.log_tail + assert SECRET not in deploy.message + for _local, remote in transport.uploads: + assert SECRET not in remote + for command in transport.commands: + assert SECRET not in command, "a credential must never reach a remote command line" + + +async def test_the_ftp_rollback_without_a_backup_is_refused_not_guessed( + tenant, session_factory +): + """FTP has no atomic switch: without the backup there is nothing to undo.""" + async with session_factory() as session: + target = await DeployTargetRepository(session).get_for_project(tenant["project_id"]) + assert target is not None + target.kind = "ftp" + await session.commit() + + transport = FakeFtpTransport() + async with session_factory() as session: + service, _tester = _service(session, ssh=FakeSshTransport(), ftp=transport) + deploy = await service.deploy_production( + organization_id=tenant["org_id"], + project_id=tenant["project_id"], + actor_user_id="u1", + confirm=tenant["slug"], + ) + await session.commit() + assert deploy.status == DeployStatus.SUCCEEDED.value, deploy.message + + # Clear the recorded backup to simulate a lost archive. + row = await ProductionDeployRepository(session).get(deploy.id) + assert row is not None + deploy_row = await session.get( + __import__( + "app.infrastructure.db.models_delivery", fromlist=["ProductionDeployRow"] + ).ProductionDeployRow, + deploy.id, + ) + assert deploy_row is not None + deploy_row.backup_path = "" + await session.commit() + + rolled = await service.rollback_production( + organization_id=tenant["org_id"], + project_id=tenant["project_id"], + actor_user_id="u1", + confirm=tenant["slug"], + deploy_id=deploy.id, + ) + await session.commit() + + assert rolled.status == DeployStatus.FAILED.value + assert "پشتیبان" in rolled.message + + +# --------------------------------------------------------------------------- +# HTTP surface +# --------------------------------------------------------------------------- + + +async def test_a_member_cannot_deploy_to_production(client: AsyncClient, tenant): + member_token = (await signup(client, "deploy-member@example.com"))["access_token"] + member_headers = auth_header(member_token) + # Join the org as a viewer so membership succeeds but the capability does not. + invite = await client.post( + f"/api/organizations/{tenant['org_id']}/invites", + json={"email": "deploy-member@example.com", "role": "viewer"}, + headers=tenant["headers"], + ) + if invite.status_code in (200, 201): + token = invite.json().get("token") or invite.json().get("invite_url", "").split("=")[-1] + if token: + await client.post( + "/api/invites/accept", + json={"token": token, "password": "S3cure-passw0rd!"}, + headers=member_headers, + ) + + response = await client.post( + f"/api/organizations/{tenant['org_id']}/delivery/production/deploy", + json={ + "project_id": tenant["project_id"], + "confirm": tenant["slug"], + }, + headers=member_headers, + ) + # Either not a member (404, the tenancy check) or a member without the + # capability (403). Never a success. + assert response.status_code in (403, 404), response.text + + +async def test_the_confirm_slug_is_enforced_over_http(client: AsyncClient, tenant): + response = await client.post( + f"/api/organizations/{tenant['org_id']}/delivery/production/deploy", + json={"project_id": tenant["project_id"], "confirm": "wrong-slug"}, + headers=tenant["headers"], + ) + assert response.status_code == 422, response.text + assert "تأییدیه" in response.json()["error"]["message"] + + +async def test_the_target_summary_never_returns_the_secret(client: AsyncClient, tenant): + response = await client.get( + f"/api/organizations/{tenant['org_id']}/delivery/production", + params={"project_id": tenant["project_id"]}, + headers=tenant["headers"], + ) + + assert response.status_code == 200, response.text + body = response.json() + assert body["target"]["has_secret"] is True + assert SECRET not in response.text + assert "secret" not in set(body["target"]), "the raw secret column must not be exposed" + + +async def test_deploying_without_a_target_is_refused(client: AsyncClient, tenant, session_factory): + async with session_factory() as session: + target = await DeployTargetRepository(session).get_for_project(tenant["project_id"]) + assert target is not None + target.is_active = False + await session.commit() + + response = await client.post( + f"/api/organizations/{tenant['org_id']}/delivery/production/deploy", + json={"project_id": tenant["project_id"], "confirm": tenant["slug"]}, + headers=tenant["headers"], + ) + assert response.status_code == 422, response.text + assert "مقصد" in response.json()["error"]["message"] diff --git a/docs/plan/reports/agent-3-phase-5.md b/docs/plan/reports/agent-3-phase-5.md new file mode 100644 index 0000000..ee778ab --- /dev/null +++ b/docs/plan/reports/agent-3-phase-5.md @@ -0,0 +1,146 @@ +# Agent 3 — Phase 5 report: production deploy (SSH / SFTP / FTP / FTPS) + +Date: 2026-10-14 · Branch: `dev` + +## What was done + +### 5.1 `adapters/deploy_ssh.py` — `SshDeployer` + +SSH and SFTP share one implementation (both are asyncssh; the difference is only which +subsystem is used). + +* **`test_connection`** — SSRF check first, then connect (15 s), capture the host key + fingerprint, and probe that `remote_path` exists and is writable. +* **Host key TOFU** — the first successful test records the fingerprint; every later + connection must match it. A mismatch is a hard failure ("کلید میزبان تغییر کرده است"), + because it is either a rebuilt server or an interception and only a human can tell. +* **`deploy`** — upload the artifact to `remote_path/releases/-/`, + extract, run `post_deploy_command` with `cwd` there (600 s, streamed), then switch the + `current` symlink atomically (`ln -sfn` to a relative target + `mv -T` on the same + filesystem), then prune to the newest 5. +* **`rollback`** — re-point `current` at the previous release (`ls -1t` is newest-first, + so it is the second entry) and re-run the hook. A rollback that rebuilt would not be a + rollback. + +**A failed deploy never moves `current`.** If the extraction or the post-deploy command +fails, the function returns before the switch, so the previous release keeps serving — +the same property the demo pipeline has, for the same reason. + +The remote scripts are built from constants, never from user input, and every argument is +`shlex.quote`d. + +### 5.2 `adapters/deploy_ftp.py` — `FtpDeployer` + +* **`test_connection`** — SSRF check, login (explicit TLS for `ftps`), `cwd` the target + path, write and delete a probe file. +* **`deploy`** — **backs up first, always**: every file about to be overwritten is + downloaded into `ARTIFACTS_ROOT///ftp-backup--.tar.gz` before + anything is uploaded. The artifact is extracted to a scratch directory first, then the + deployable build is chosen (`dist/`, `build/`, `public/`, `out/`, else the root) — a + static deploy must not publish the repository. +* **`rollback`** — re-upload that archive. + +FTP's limits are documented rather than hidden: no atomic switch, no host key to pin, and +without the backup there is nothing to undo. `rollback_production` refuses with a plain +message in that case rather than guessing. + +**Archive safety.** `_safe_member` sanitises every tar member before extraction: absolute +names and `..` are refused, as are symlink/hardlink members whose target escapes. An +uploaded artifact is not trusted, and `filter="data"` alone is not enough (it does not +skip the archive's own root entry, which is the member that broke extraction here). + +### 5.3 `adapters/deploy_target.py` — `DispatchingDeployTargetTester` + +Routes `test_connection` on `spec.kind` and exposes `deployer_for(kind)`. The wizard sees +one shape; the differences between SSH and FTP stay inside the deployers. + +### 5.4 `services/deploy_service.py` + endpoints + +`production_deploys` table (migration `a3b4c5d6e7f4`, chained off this workstream's own +head so the three branches still merge cleanly). + +| Method | Path | Role | +|---|---|---| +| `GET` | `/delivery/production` | admin+ — target summary + deploy history | +| `POST` | `/delivery/production/test` | admin+ — test the target, record the result | +| `POST` | `/delivery/production/deploy` | org owner/admin, slug confirmed, audited | +| `POST` | `/delivery/production/rollback` | org owner/admin, slug confirmed, audited | + +**One production deploy per project at a time.** Two concurrent deploys to the same target +interleave their writes on a shared host — over SSH merely confusing, over FTP how a site +ends up in a state nobody can describe. + +**There is no code path from a pipeline run into this service.** The engine's furthest +reach is "merged and pushed to `dev`"; production needs a human, a typed confirmation and +an audit row. That is D13, and it is enforced by the module boundary, not by a flag. + +The target summary returns `has_secret: true`, never the secret. `_spec_from_target` is the +only place a secret is decrypted, and the plaintext lives for the duration of one call. + +## Files changed + +| File | Change | +|---|---| +| `app/infrastructure/adapters/deploy_ssh.py` | new — `SshDeployer`, `AsyncSshTransport` | +| `app/infrastructure/adapters/deploy_ftp.py` | new — `FtpDeployer`, `AsyncFtpTransport`, safe extraction | +| `app/infrastructure/adapters/deploy_target.py` | new — the dispatching tester | +| `app/application/services/deploy_service.py` | new — the deploy use cases | +| `app/domain/delivery/models.py` | `DeployStatus`, `ProductionDeploy` | +| `app/infrastructure/db/models_delivery.py` | `ProductionDeployRow` | +| `alembic/versions/a3b4c5d6e7f4_production_deploys.py` | new migration | +| `app/infrastructure/repositories/delivery.py` | `ProductionDeployRepository` | +| `app/infrastructure/repositories/projects.py` | `DeployTargetRepository` | +| `app/application/container_delivery.py`, `container.py`, `deps.py` | wiring | +| `app/interfaces/http/routers/delivery.py` | 4 production routes | +| `tests/test_deploy.py` | new — 26 tests | + +## Test output + +``` +apps/api: 499 passed, 1 skipped (was 473; +26) + ruff check . -> All checks passed! + mypy app -> Success: no issues found in 136 source files +alembic: upgrade head / check -> clean +``` + +## Deviations from the TODO + +1. **`post_deploy_command` is a deploy parameter, not part of `DeployTargetSpec`.** + `ports/deploy_target.py` is a frozen contract and the spec has no such field; the + service that owns the `deploy_targets` row passes it explicitly instead. No contract + change. +2. **`filter="data"` replaced by `_safe_member`.** The stdlib filter does not skip the + archive's root member, which made extraction fail; the custom filter is stricter + (it also refuses escaping link targets) and is the honest implementation of "an + uploaded artifact is not trusted". +3. **`a3b4c5d6e7f4` is a second migration for this workstream.** The plan allows it + ("more only if chained off your own"); chaining off `a3b4c5d6e7f3` keeps the merge with + Agent 1 and Agent 2 trivial. +4. **The `artifact` for FTP is extracted before the build directory is chosen.** The + artifact is a tar.gz of the whole release, not a directory; the TODO's "the build dir + is set by the framework catalog default or `dist/`, `build/`, `public/`" describes the + choice, not the container. + +## Open questions / CHANGE REQUESTs + +- **CHANGE REQUEST (for integration):** `DeployTargetSpec` would benefit from + `post_deploy_command` and `organization_id`/`project_id`. Both are currently passed as + extra keyword arguments, which works but means the spec is not the whole story of a + deploy. No change is needed for this phase to be correct; noting it so the contract can + be widened deliberately at integration rather than by accident. +- **CHANGE REQUEST (for integration):** `build_deploy_target_tester` now returns the real + `DispatchingDeployTargetTester`, so Agent 1's wizard will attempt a real connection. + Its tests must inject a fake tester, as `tests/test_deploy.py` does. + +## Known limitations + +- `AsyncSshTransport` passes `known_hosts=None` and enforces the pinned key by comparing + the fingerprint after connecting. That is TOFU, but it means the first byte of the + handshake is accepted before the check; a pre-seeded `known_hosts` would be stricter. + The fingerprint check happens before any command runs or any file is uploaded. +- FTPS does not verify the server certificate (`CERT_NONE`): a shared host's FTP + certificate is rarely matchable to the domain the operator typed. This is documented in + the adapter and stated in the test message the operator sees. +- No `@pytest.mark.deploy_it` integration test against a real sshd/pure-ftpd yet; the fake + transports cover the mechanics, and a live round-trip needs infrastructure the repo + cannot assume.