diff --git a/.github/workflows/deploy-digitalocean.yml b/.github/workflows/deploy-digitalocean.yml new file mode 100644 index 000000000..8cb2fc915 --- /dev/null +++ b/.github/workflows/deploy-digitalocean.yml @@ -0,0 +1,72 @@ +name: Deploy workspace backend to DigitalOcean + +on: + push: + branches: [release] + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: workspace-backend-production + cancel-in-progress: false + +jobs: + build: + if: github.ref == 'refs/heads/release' + runs-on: ubuntu-latest + timeout-minutes: 20 + steps: + - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 + with: + persist-credentials: false + - name: Test deployment guards + run: python3 -m unittest discover -s workspace/operations/digitalocean/cd/tests -v + - name: Build exact release without deployment credentials + env: + RELEASE_SHA: ${{ github.sha }} + DOCKER_BUILDKIT: "1" + run: | + set -euo pipefail + # Canary files must be excluded even inside copied application paths. + mkdir -p workspace/backend/app/secrets + printf "not-a-real-secret" > workspace/backend/app/.env.deploy-canary + printf "not-a-real-secret" > workspace/backend/app/secrets/deploy-canary + docker build --file workspace/backend/production.Dockerfile \ + --build-arg RELEASE_SHA="$RELEASE_SHA" \ + --tag "openagents-backend:release-$RELEASE_SHA" workspace/backend + docker run --rm --network none --entrypoint python \ + -e DATABASE_URL=sqlite:// -e BACKGROUND_JOBS_ENABLED=false \ + "openagents-backend:release-$RELEASE_SHA" \ + -c 'import app.main; import app.cache; import yaml; from pathlib import Path; assert not Path("/app/app/.env.deploy-canary").exists(); assert not Path("/app/app/secrets/deploy-canary").exists()' + docker save "openagents-backend:release-$RELEASE_SHA" | gzip -1 > backend-image.tar.gz + - uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4 + with: + name: backend-image-${{ github.sha }} + path: backend-image.tar.gz + retention-days: 3 + compression-level: 0 + if-no-files-found: error + + deploy: + needs: build + runs-on: ubuntu-latest + environment: digitalocean-production + timeout-minutes: 45 + steps: + - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 + with: + persist-credentials: false + - uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093 # v4 + with: + name: backend-image-${{ github.sha }} + - name: Stage both nodes and request a rolling deployment + env: + RELEASE_SHA: ${{ github.sha }} + DO_DEPLOY_SSH_KEY: ${{ secrets.DO_DEPLOY_SSH_KEY }} + DO_DEPLOY_KNOWN_HOSTS: ${{ secrets.DO_DEPLOY_KNOWN_HOSTS }} + DO_PRIMARY_HOST: ${{ vars.DO_PRIMARY_HOST }} + DO_SECONDARY_HOST: ${{ vars.DO_SECONDARY_HOST }} + GH_TOKEN: ${{ github.token }} + run: python3 workspace/operations/digitalocean/cd/ci.py diff --git a/.github/workflows/promote-release.yml b/.github/workflows/promote-release.yml index c7dd646f1..5a89abf88 100644 --- a/.github/workflows/promote-release.yml +++ b/.github/workflows/promote-release.yml @@ -2,7 +2,7 @@ name: Promote develop to release # The only sanctioned way to move `release`. # -# `release` is what Railway deploys. It must always be a fast-forward of +# `release` triggers the DigitalOcean backend deployment and Vercel frontend deployment. It must always be a fast-forward of # `develop`: never a merge commit, never a commit that isn't on develop. A # ruleset on `release` blocks direct pushes from everyone; this workflow pushes # with a dedicated deploy key (RELEASE_PROMOTE_SSH_KEY) that the ruleset lets diff --git a/.gitignore b/.gitignore index 73002842c..d05476c74 100644 --- a/.gitignore +++ b/.gitignore @@ -280,3 +280,6 @@ workspace/backend/secrets/ # End-to-end test credentials (tests/end_to_end/agents.example.json is the template) tests/end_to_end/agents.config.json + +# Reviewed production dependency constraints +!workspace/backend/requirements.production.lock diff --git a/workspace/backend/production.Dockerfile b/workspace/backend/production.Dockerfile new file mode 100644 index 000000000..86c8855a7 --- /dev/null +++ b/workspace/backend/production.Dockerfile @@ -0,0 +1,19 @@ +FROM python:3.12-slim +WORKDIR /app +RUN apt-get update && apt-get install -y --no-install-recommends libpq5 \ + && rm -rf /var/lib/apt/lists/* +COPY requirements.txt requirements.production.lock ./ +RUN pip install --no-cache-dir -r requirements.txt -c requirements.production.lock 'PyYAML>=6.0' +# Explicit runtime allowlist: no credentials, tests, Git history or operations files. +COPY app/ ./app/ +COPY alembic/ ./alembic/ +COPY openagents/ ./openagents/ +COPY cloud_providers/ ./cloud_providers/ +COPY registry/ ./registry/ +COPY alembic.ini entrypoint.sh ./ +RUN chmod +x entrypoint.sh +ARG RELEASE_SHA +LABEL org.opencontainers.image.revision=$RELEASE_SHA +ENV PYTHONPATH=/app PYTHONUNBUFFERED=1 +EXPOSE 8000 +ENTRYPOINT ["./entrypoint.sh"] diff --git a/workspace/backend/production.Dockerfile.dockerignore b/workspace/backend/production.Dockerfile.dockerignore new file mode 100644 index 000000000..dcdbf1580 --- /dev/null +++ b/workspace/backend/production.Dockerfile.dockerignore @@ -0,0 +1,23 @@ +** +!production.Dockerfile +!requirements.txt +!requirements.production.lock +!entrypoint.sh +!alembic.ini +!app/ +!app/** +!alembic/ +!alembic/** +!openagents/ +!openagents/** +!cloud_providers/ +!cloud_providers/** +!registry/ +!registry/** +**/__pycache__/** +**/*.pyc +**/.env* +**/secrets/** +**/*.pem +**/*.key +**/*.p8 diff --git a/workspace/backend/requirements.production.lock b/workspace/backend/requirements.production.lock new file mode 100644 index 000000000..2f5a2c919 --- /dev/null +++ b/workspace/backend/requirements.production.lock @@ -0,0 +1,75 @@ +alembic==1.20.0 +annotated-doc==0.0.5 +annotated-types==0.8.0 +anthropic==1.13.0 +anyio==4.15.1 +boto3==1.43.111 +botocore==1.43.111 +CacheControl==0.14.4 +certifi==2026.7.22 +cffi==2.1.1 +charset-normalizer==3.5.2 +click==8.5.0 +cryptography==50.0.2 +docstring_parser==0.18.0 +fastapi==0.143.0 +firebase_admin==7.7.0 +google-api-core==2.42.0 +google-auth==2.61.0 +google-cloud-core==2.8.0 +google-cloud-firestore==2.34.1 +google-cloud-storage==3.17.0 +google-crc32c==1.9.0 +google-resumable-media==2.11.0 +googleapis-common-protos==1.75.5 +greenlet==3.5.6 +grpcio==1.84.0 +grpcio-status==1.84.0 +h11==0.16.0 +h2==4.4.1 +hpack==4.2.0 +httpcore==1.0.9 +httpcore2==2.13.1 +httptools==0.9.0 +httpx==0.28.1 +httpx2==2.13.1 +hyperframe==6.1.0 +idna==3.20 +jiter==0.17.0 +jmespath==1.1.0 +Mako==1.4.3 +MarkupSafe==3.0.4 +msgpack==1.2.3 +openai==3.28.0 +opentelemetry-api==1.45.1 +pillow==12.3.0 +playwright==1.63.0 +proto-plus==1.29.0 +protobuf==7.36.2 +psycopg2-binary==2.9.13 +pyasn1==0.6.4 +pyasn1_modules==0.4.2 +pycparser==3.11 +pydantic==2.14.0 +pydantic_core==2.50.0 +pyee==13.0.1 +PyJWT==2.15.1 +python-dateutil==2.9.0.post0 +python-dotenv==1.2.4 +python-multipart==0.0.32 +PyYAML==6.0.3 +redis==8.1.0 +requests==2.34.2 +s3transfer==0.19.2 +six==1.17.0 +sniffio==1.3.1 +SQLAlchemy==2.0.54 +starlette==1.7.0 +truststore==0.10.4 +typing-inspection==0.4.4 +typing_extensions==4.16.0 +urllib3==2.8.0 +uvicorn==0.54.0 +uvloop==0.23.0 +watchfiles==1.3.0 +websockets==17.2 diff --git a/workspace/operations/digitalocean/cd/README.md b/workspace/operations/digitalocean/cd/README.md new file mode 100644 index 000000000..778f4c615 --- /dev/null +++ b/workspace/operations/digitalocean/cd/README.md @@ -0,0 +1,70 @@ +# DigitalOcean automatic backend deployments + +A push to `release` builds that exact commit and rolls it out to the secondary, +then the primary backend. Use the existing **Promote develop to release** workflow +to ship a merged develop commit. The frontend continues using its existing deployment. +The deployment workflow can also be run manually against `release` to retry it. + +The build job receives no deployment secrets. Its explicit Docker context allowlist +excludes environment files, operations files and credentials. The deployment job uses +the `digitalocean-production` GitHub environment, restricted to `release`, with: + +- Secrets `DO_DEPLOY_SSH_KEY`, `DO_DEPLOY_KNOWN_HOSTS`. +- Variables `DO_PRIMARY_HOST`, `DO_SECONDARY_HOST`. + +The dedicated SSH key is restricted by root-owned forced commands on both servers. +It can stage an image, start a rollout, and inspect deployment status, but cannot +open an interactive shell or change routing directly. A separate restricted key +allows the coordinator to update the secondary. Host keys are pinned. Anyone able +to publish release code can execute application code in production; branch protections +remain essential. Do not copy developer SSH keys into GitHub. + +## Server installation + +Install `deploy.py`, `rollout.py`, `migration_plan.py`, and `backup.py` under the +root-owned `/opt/openagents/cd` directory (mode 0700). Python dependencies are PyYAML, +boto3 and botocore; PostgreSQL client tools must support the server version. +Root-owned `config.json` supplies coordinator flag, node IDs, peer host, local/peer +Nginx upstream addresses, upstream file, public URL, load balancer ID, DO token file +and migration Docker network. Node configuration and credentials are not tracked. + +Keep all credentials in `cd/secrets` (0700 directory, 0600 files): `migration.env` +(admin DATABASE_URL), `smoke.json` (workspace and token), `backup.json` (libpq env and +Spaces settings), and the coordinator's `peer-key`. The application's existing +`/opt/openagents/secrets/production.env` stays on each server. Pin peer SSH host keys +in `cd/peer-known-hosts`. Configure authorized_keys with `restrict` and a forced +`python3 /opt/openagents/cd/deploy.py ci` or `peer` command; restrict the peer key's +source address to the coordinator's private IP. + +## Rollout and migrations + +Both nodes must be ready, contain the candidate image, and be the exact members of +the load balancer. The coordinator compares database revision to the candidate's +Alembic head. Every pending migration must explicitly declare +`deployment_backward_compatible = True`; review this declaration carefully. Use +expand/contract changes compatible with the currently running image. Unmarked +migrations stop deployment before any traffic changes. The declaration is a review +contract, not a proof of compatibility. + +Pending migrations trigger a consistent PostgreSQL custom-format backup in the +private Spaces bucket. The archive is checked with pg_restore, downloaded and +SHA-256 verified, and anonymous access must be denied. Migration runs once, before +replacing either node. Database schema/data are never automatically downgraded. + +For each node, remove it from the load balancer and temporarily forward new requests +to its peer, update its image, check readiness/cache/authenticated reads, restore local +routing and re-add it. Long-lived connections on the replaced process may reconnect. +The primary retains background jobs; the secondary has them disabled. This does not +provide automatic scheduler failover. Public verification follows the rollout. +Failures restore previous images; failed rollback nodes remain out of rotation. + +The coordinator runs as `openagents-auto-deploy.service`, independently of Actions +connectivity. `cd/result-.json` and `cd/last-result.json` record status. Detailed +logs stay in root-only `cd/operations.log`; they may contain sensitive application +errors and must not be pasted into GitHub. If status is `interrupted` or +`rollback_failed`, inspect node readiness, routing and load balancer membership +before retrying. Do not automatically restart a partially completed rollout. + +The forced-command scripts are host administration code, not automatically replaced +by application releases. Changes to them require testing and explicit installation. +Run tests with `python3 -m unittest discover -s workspace/operations/digitalocean/cd/tests -v`. diff --git a/workspace/operations/digitalocean/cd/backup.py b/workspace/operations/digitalocean/cd/backup.py new file mode 100644 index 000000000..9a1c57dfb --- /dev/null +++ b/workspace/operations/digitalocean/cd/backup.py @@ -0,0 +1,56 @@ +#!/usr/bin/env python3 +"""Consistent pre-migration PostgreSQL snapshot, privately verified in Spaces.""" +import datetime +import hashlib +import json +import os +from pathlib import Path +import subprocess +import urllib.request +import urllib.error +import boto3 +from botocore.config import Config + +ROOT = Path('/opt/openagents/cd') + + +def backup(): + cfg = json.loads((ROOT / 'secrets/backup.json').read_text()) + stamp = datetime.datetime.now(datetime.timezone.utc).strftime('%Y%m%dT%H%M%SZ') + directory = ROOT / 'backups'; directory.mkdir(mode=0o700, exist_ok=True) + path = directory / ('pre-migration-' + stamp + '.dump') + env = {**os.environ, **cfg['pg_env']} + # libpq credentials are inherited privately, never command-line arguments. + subprocess.run(['pg_dump', '--format=custom', '--no-owner', '--no-acl', + '--file', str(path)], env=env, check=True, timeout=1200) + subprocess.run(['pg_restore', '--list', str(path)], stdout=subprocess.DEVNULL, check=True) + with path.open('rb') as f: digest = hashlib.file_digest(f, 'sha256').hexdigest() + s3 = boto3.client('s3', endpoint_url=cfg['endpoint'], region_name=cfg['region'], + aws_access_key_id=cfg['key_id'], aws_secret_access_key=cfg['secret'], + config=Config(signature_version='s3v4')) + key = 'database-backups/automatic-deploy/' + path.name + s3.upload_file(str(path), cfg['bucket'], key, + ExtraArgs={'ACL': 'private', 'Metadata': {'sha256': digest}}) + head = s3.head_object(Bucket=cfg['bucket'], Key=key) + if head['ContentLength'] != path.stat().st_size or head['Metadata'].get('sha256') != digest: + raise RuntimeError('Backup object metadata mismatch') + body = s3.get_object(Bucket=cfg['bucket'], Key=key)['Body'] + verify = hashlib.sha256() + with body: + while chunk := body.read(8 * 1024 * 1024): verify.update(chunk) + if verify.hexdigest() != digest: raise RuntimeError('Backup checksum mismatch') + try: + with urllib.request.urlopen(urllib.request.Request( + cfg['public_bucket_url'] + '/' + key, method='HEAD'), timeout=15): + raise RuntimeError('Backup is publicly accessible') + except urllib.error.HTTPError as exc: + if exc.code not in (403, 404): raise + receipt = {'key': key, 'bytes': path.stat().st_size, 'sha256': digest, + 'download_verified': True, 'anonymous_access_denied': True} + path.with_suffix('.json').write_text(json.dumps(receipt, indent=2)) + print(json.dumps(receipt)) + + +if __name__ == '__main__': + os.umask(0o077) + backup() diff --git a/workspace/operations/digitalocean/cd/ci.py b/workspace/operations/digitalocean/cd/ci.py new file mode 100644 index 000000000..13b2a2733 --- /dev/null +++ b/workspace/operations/digitalocean/cd/ci.py @@ -0,0 +1,56 @@ +#!/usr/bin/env python3 +"""Actions client. SSH keys never enter the checkout, image or build job.""" +import json +import os +from pathlib import Path +import re +import subprocess +import tempfile +import time + + +def main(): + sha = os.environ['RELEASE_SHA'] + if not re.fullmatch('[0-9a-f]{40}', sha): raise ValueError('Invalid release') + hosts = [os.environ['DO_PRIMARY_HOST'], os.environ['DO_SECONDARY_HOST']] + if any(not re.fullmatch('[0-9.]+', host) for host in hosts): + raise ValueError('Invalid deployment host') + current = subprocess.check_output(['gh', 'api', + 'repos/' + os.environ['GITHUB_REPOSITORY'] + '/git/ref/heads/release', + '--jq', '.object.sha'], text=True).strip() + if current != sha: + print('Skipping superseded release'); return + os.umask(0o077) + with tempfile.TemporaryDirectory(prefix='backend-deploy-') as directory: + key, known = Path(directory) / 'key', Path(directory) / 'known_hosts' + key.write_text(os.environ.pop('DO_DEPLOY_SSH_KEY').strip() + '\n') + known.write_text(os.environ.pop('DO_DEPLOY_KNOWN_HOSTS').strip() + '\n') + options = ['ssh', '-i', str(key), '-o', 'IdentitiesOnly=yes', '-o', 'BatchMode=yes', + '-o', 'StrictHostKeyChecking=yes', '-o', 'UserKnownHostsFile=' + str(known), + '-o', 'ConnectTimeout=15', '-o', 'ServerAliveInterval=15', + '-o', 'ServerAliveCountMax=4'] + def rpc(host, command, source=None, timeout=900): + output = subprocess.check_output(options + ['root@' + host, command], + stdin=source, timeout=timeout) + return json.loads(output) + for host in hosts: + with open('backend-image.tar.gz', 'rb') as archive: + rpc(host, 'stage ' + sha, archive) + print('Image staged on a backend', flush=True) + rpc(hosts[0], 'deploy ' + sha) + for _ in range(160): + time.sleep(15) + try: + result = rpc(hosts[0], 'result ' + sha, timeout=45) + except (subprocess.SubprocessError, json.JSONDecodeError): + continue # server rollout survives a transient SSH interruption + if result['status'] == 'running': continue + print('Deployment result: ' + result['status'], flush=True) + if result['status'] != 'deployed': + raise RuntimeError('Deployment did not succeed; inspect private server logs') + return + raise RuntimeError('Timed out waiting; check server state before retrying') + + +if __name__ == '__main__': + main() diff --git a/workspace/operations/digitalocean/cd/deploy.py b/workspace/operations/digitalocean/cd/deploy.py new file mode 100644 index 000000000..dc2a85564 --- /dev/null +++ b/workspace/operations/digitalocean/cd/deploy.py @@ -0,0 +1,314 @@ +#!/usr/bin/env python3 +"""Root-owned forced SSH command; all credentials/configuration stay on the host.""" +import contextlib +import fcntl +import gzip +import json +import os +from pathlib import Path +import re +import shlex +import shutil +import subprocess +import sys +import time +import urllib.request +import urllib.error +from rollout import rollout + +ROOT = Path('/opt/openagents') +CD = ROOT / 'cd' +SHA = re.compile(r'^[0-9a-f]{40}$') +IMAGE = re.compile(r'^openagents-backend:(?:release-[0-9a-f]{40}|v11-[0-9a-f]{9})$') + + +def config(): + return json.loads((CD / 'config.json').read_text()) + + +def write_private(path, value): + tmp = path.with_suffix(path.suffix + '.tmp') + tmp.write_text(value) + tmp.chmod(0o600) + tmp.replace(path) + + +def command(args, timeout=300, input=None, capture=False): + # Output may contain connection strings or migration parameters. It stays + # in a root-only server log and never travels to Actions on a failure. + result = subprocess.run(args, cwd=ROOT, input=input, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, timeout=timeout) + with (CD / 'operations.log').open('ab') as log: + log.write(result.stdout); log.write(result.stderr) + if result.returncode: + raise RuntimeError('Server command failed; inspect private deployment log') + return result.stdout.decode() if capture else None + + +@contextlib.contextmanager +def lock(name): + with (CD / (name + '.lock')).open('w') as file: + fcntl.flock(file, fcntl.LOCK_EX | fcntl.LOCK_NB) + yield + + +def ready(url='http://127.0.0.1:8000/ready'): + try: + req = urllib.request.Request(url, headers={'User-Agent': 'curl/8.0'}) + with urllib.request.urlopen(req, timeout=5) as response: + return response.status == 200 + except Exception: + return False + + +def status(): + image = command(['docker', 'inspect', '-f', '{{.Config.Image}}', + 'openagents-backend-1'], capture=True).strip() + return {'image': image, 'ready': ready()} + + +def validate(image): + if not IMAGE.fullmatch(image): + raise ValueError('Invalid image reference') + revision = command(['docker', 'image', 'inspect', '-f', + '{{index .Config.Labels "org.opencontainers.image.revision"}}', + image], capture=True).strip() + if ':release-' in image and revision != image.split(':release-')[1]: + raise ValueError('Image revision does not match release') + + +def stage(sha): + if not SHA.fullmatch(sha): + raise ValueError('Invalid commit') + image = 'openagents-backend:release-' + sha + with lock('stage'): + # Gzip is decoded into a private temporary archive with a size limit. + # Docker handles its own image format; no shell or arbitrary tar extraction. + archive = CD / 'incoming-image.tar' + try: + with gzip.GzipFile(fileobj=sys.stdin.buffer) as source, archive.open('wb') as target: + total = 0 + while chunk := source.read(1024 * 1024): + total += len(chunk) + if total > 3 * 1024**3: + raise ValueError('Image archive exceeds limit') + target.write(chunk) + command(['docker', 'load', '-i', str(archive)], timeout=600) + validate(image) + finally: + archive.unlink(missing_ok=True) + return {'staged': sha} + + +def route(mode): + cfg = config() + address = cfg['local_upstream'] if mode == 'local' else cfg['peer_upstream'] + if mode not in ('local', 'peer') or not re.fullmatch(r'[0-9.:]+', address): + raise ValueError('Invalid routing target') + path = Path(cfg['upstream_file']); old = path.read_text() + new = re.sub(r'server\s+[0-9.:]+;', 'server ' + address + ';', old) + if new == old: + return + path.write_text(new) + try: + command(['nginx', '-t']) + command(['systemctl', 'reload', 'nginx']) + except Exception: + path.write_text(old) + raise + + +def activate(image): + import yaml + validate(image) + with lock('activate'): + path = ROOT / 'compose.yml' + old = path.read_text() + write_private(CD / 'compose-before.yml', old) + data = yaml.safe_load(old) + data['services']['backend']['image'] = image + write_private(path, yaml.safe_dump(data, sort_keys=False)) + command(['docker', 'compose', '--profile', 'production', 'up', '-d', + '--no-deps', '--timeout', '25', 'backend'], timeout=180) + deadline = time.monotonic() + 120 + consecutive = 0 + while time.monotonic() < deadline: + consecutive = consecutive + 1 if ready() else 0 + if consecutive >= 3: + return status() + time.sleep(2) + raise RuntimeError('Candidate readiness failed') + + +def verify(public=False): + cfg = config() + base = cfg['public_url'] if public else 'http://127.0.0.1:8000' + if not ready(base + '/ready'): + raise RuntimeError('Readiness failed') + if not public: + # Treat shared-cache failure as a failed deployment, even though the + # application can fall back to SQL during an ordinary cache outage. + command(['docker', 'exec', 'openagents-backend-1', 'python', '-c', + 'from app import cache; c=cache._lazy_client(); assert c is not None and c.ping()']) + smoke = json.loads((CD / 'secrets/smoke.json').read_text()) + from urllib.parse import urlencode + query = urlencode({'network': smoke['workspace'], 'limit': 1}) + req = urllib.request.Request(base + '/v1/events?' + query, + headers={'User-Agent': 'curl/8.0', 'X-Workspace-Token': smoke['token']}) + with urllib.request.urlopen(req, timeout=15) as response: + body = json.load(response) + if response.status != 200 or not isinstance(body.get('data', {}).get('events'), list): + raise RuntimeError('Authenticated smoke check failed') + return {'verified': True} + + +class Node: + def __init__(self, name, cfg, remote=False): + self.name, self.cfg, self.remote = name, cfg, remote + self.id = cfg['peer_droplet_id'] if remote else cfg['droplet_id'] + + def rpc(self, verb, arg=None): + if not self.remote: + return dispatch('peer', [verb] + ([arg] if arg else [])) + args = ['ssh', '-i', str(CD / 'secrets/peer-key'), '-o', 'BatchMode=yes', + '-o', 'StrictHostKeyChecking=yes', '-o', 'UserKnownHostsFile=' + str(CD / 'peer-known-hosts'), + '-o', 'ConnectTimeout=10', 'root@' + self.cfg['peer_host'], + ' '.join([verb] + ([arg] if arg else []))] + return json.loads(command(args, capture=True, timeout=300)) + + def status(self): return self.rpc('status') + def validate(self, image): return self.rpc('validate', image) + def activate(self, image): return self.rpc('activate', image) + def route(self, mode): return self.rpc('route', mode) + def verify(self): return self.rpc('verify') + + +class Balancer: + def __init__(self, cfg): self.cfg = cfg + + def api(self, suffix='', method='GET', body=None): + token = Path(self.cfg['do_token_file']).read_text().strip() + req = urllib.request.Request('https://api.digitalocean.com/v2/load_balancers/' + self.cfg['load_balancer_id'] + suffix, + method=method, headers={'Authorization': 'Bearer ' + token, 'Content-Type': 'application/json'}, + data=json.dumps(body).encode() if body is not None else None) + with urllib.request.urlopen(req, timeout=30) as response: + raw = response.read() + return json.loads(raw) if raw else {} + + def contains_all(self, nodes): + actual = set(self.api()['load_balancer']['droplet_ids']) + return actual == {n.id for n in nodes} + + def remove(self, node): + ids = self.api()['load_balancer']['droplet_ids'] + if node.id in ids: + if len(ids) < 2: + raise RuntimeError('Refusing to remove last backend') + self.api('/droplets', 'DELETE', {'droplet_ids': [node.id]}) + + def add(self, node): + if node.id not in self.api()['load_balancer']['droplet_ids']: + self.api('/droplets', 'POST', {'droplet_ids': [node.id]}) + + +def migration_command(image, *args): + return ['docker', 'run', '--rm', '--network', config()['migration_network'], + '--env-file', str(CD / 'secrets/migration.env'), + '-v', str(CD / 'migration_plan.py') + ':/ops/migration_plan.py:ro', + '--entrypoint', 'python', image, *args] + + +def migrate(image): + plan = json.loads(command(migration_command(image, '/ops/migration_plan.py'), capture=True)) + if plan['pending']: + command(['python3', str(CD / 'backup.py')], timeout=1800) + command(migration_command(image, '-m', 'alembic', 'upgrade', 'head'), timeout=900) + after = json.loads(command(migration_command(image, '/ops/migration_plan.py'), capture=True)) + if after['pending'] or after['current'] != plan['target']: + raise RuntimeError('Schema did not reach target revision') + write_private(CD / 'migration-last.json', json.dumps(after)) + + +def deploy(sha): + if not SHA.fullmatch(sha): raise ValueError('Invalid commit') + cfg = config() + if not cfg['coordinator']: raise ValueError('Deploy requires coordinator') + with lock('deployment'): + image = 'openagents-backend:release-' + sha + nodes = [Node('secondary', cfg, True), Node('primary', cfg)] + def record(result): + result.update(release=sha, timestamp=time.time()) + write_private(CD / 'last-result.json', json.dumps(result)) + write_private(CD / ('result-' + sha + '.json'), json.dumps(result)) + rollout(nodes, Balancer(cfg), image, lambda: migrate(image), + lambda: verify(public=True), record, + lambda phase: time.sleep(15 if phase == 'drain' else 35)) + return {'deployed': sha} + + +def start_deploy(sha): + if not SHA.fullmatch(sha) or not config()['coordinator']: + raise ValueError('Invalid coordinator deployment') + with lock('dispatch'): + active = subprocess.run(['systemctl', 'is-active', '--quiet', 'openagents-auto-deploy.service']) + if active.returncode == 0: + raise RuntimeError('A deployment is already running') + write_private(CD / ('result-' + sha + '.json'), json.dumps({'status': 'running', 'release': sha})) + command(['systemd-run', '--unit=openagents-auto-deploy', '--collect', + '--property=RuntimeMaxSec=3600', + '--property=StandardOutput=append:' + str(CD / 'operations.log'), + '--property=StandardError=append:' + str(CD / 'operations.log'), + sys.executable, str(CD / 'deploy.py'), 'run', sha]) + return {'status': 'running', 'release': sha} + + +def deployment_result(sha): + if not SHA.fullmatch(sha): raise ValueError('Invalid commit') + result = json.loads((CD / ('result-' + sha + '.json')).read_text()) + if result['status'] == 'running': + active = subprocess.run(['systemctl', 'is-active', '--quiet', 'openagents-auto-deploy.service']) + if active.returncode != 0: + result['status'] = 'interrupted' + return result + + +def dispatch(role, words): + if not words: raise ValueError('Missing command') + verb, *args = words + if verb == 'status' and not args: return status() + if role == 'ci' and verb == 'stage' and len(args) == 1: return stage(args[0]) + if role == 'ci' and verb == 'deploy' and len(args) == 1: return start_deploy(args[0]) + if role == 'ci' and verb == 'result' and len(args) == 1: return deployment_result(args[0]) + if role == 'peer': + if verb == 'validate' and len(args) == 1: validate(args[0]); return {'valid': True} + if verb == 'route' and len(args) == 1: route(args[0]); return {'routed': args[0]} + if verb == 'activate' and len(args) == 1: return activate(args[0]) + if verb == 'verify' and not args: return verify() + raise ValueError('Command not allowed for this key') + + +if __name__ == '__main__': + os.umask(0o077) + try: + role = sys.argv[1] + if role == 'run' and len(sys.argv) == 3: + sha = sys.argv[2] + try: + print(json.dumps(deploy(sha)), flush=True) + except Exception: + path = CD / ('result-' + sha + '.json') + if json.loads(path.read_text())['status'] == 'running': + write_private(path, json.dumps({'status': 'failed', 'release': sha})) + raise + sys.exit(0) + if role not in ('ci', 'peer'): raise ValueError('Invalid role') + # SSH_ORIGINAL_COMMAND is untrusted; parse tokens, never run a shell. + words = shlex.split(os.environ.get('SSH_ORIGINAL_COMMAND', '')) + print(json.dumps(dispatch(role, words)), flush=True) + except Exception as exc: + import traceback + with (CD / 'operations.log').open('a') as log: + traceback.print_exc(file=log) + # No exception text/traceback in SSH output: libraries can include secrets. + print(json.dumps({'error': type(exc).__name__, 'detail': 'Deployment command failed; inspect private server logs'}), flush=True) + sys.exit(1) diff --git a/workspace/operations/digitalocean/cd/migration_plan.py b/workspace/operations/digitalocean/cd/migration_plan.py new file mode 100644 index 000000000..0e98f2ec9 --- /dev/null +++ b/workspace/operations/digitalocean/cd/migration_plan.py @@ -0,0 +1,35 @@ +"""Run inside the candidate image; print only revision metadata, never DSNs.""" +import json +import os + + +def plan(script, current): + heads = script.get_heads() + if len(heads) != 1 or len(current) != 1: + raise RuntimeError('Automatic deployment requires a single current and target revision') + pending = list(script.iterate_revisions(heads[0], current[0])) + unsafe = [r.revision for r in pending + if getattr(r.module, 'deployment_backward_compatible', False) is not True] + if unsafe: + raise RuntimeError('New migrations must explicitly declare deployment_backward_compatible = True') + return {'current': current[0], 'target': heads[0], + 'pending': [r.revision for r in reversed(pending)]} + + +def main(): + from alembic.config import Config + from alembic.script import ScriptDirectory + from alembic.runtime.migration import MigrationContext + from sqlalchemy import create_engine + script = ScriptDirectory.from_config(Config('/app/alembic.ini')) + engine = create_engine(os.environ['DATABASE_URL']) + try: + with engine.connect() as connection: + current = MigrationContext.configure(connection).get_current_heads() + print(json.dumps(plan(script, current))) + finally: + engine.dispose() + + +if __name__ == '__main__': + main() diff --git a/workspace/operations/digitalocean/cd/rollout.py b/workspace/operations/digitalocean/cd/rollout.py new file mode 100644 index 000000000..a3b072815 --- /dev/null +++ b/workspace/operations/digitalocean/cd/rollout.py @@ -0,0 +1,59 @@ +"""Rolling deployment policy. Dependencies are injected for deterministic failure tests.""" + +class RolloutError(RuntimeError): + pass + + +def rollout(nodes, balancer, image, migrate, verify_public, record, settle): + """Update secondary then primary; keep a verified peer available throughout. + + Migrations must be explicitly backward compatible. Rollback changes images, + never database schema or data. Failed rollback nodes remain out of the pool. + """ + before = {n.name: n.status() for n in nodes} + for n in nodes: + if not before[n.name]['ready']: + raise RolloutError('Both nodes must be ready before deployment') + n.validate(image) + if not balancer.contains_all(nodes): + raise RolloutError('Expected both backend nodes in the load balancer') + migrate() + changed = [] + try: + for n in nodes: + peer = next(other for other in nodes if other is not n) + if not peer.status()['ready']: + raise RolloutError('Peer is not ready; refusing to drain node') + changed.append(n) # include partial failures during drain/activation + balancer.remove(n) + n.route('peer') + settle('drain') + n.activate(image) + n.verify() + n.route('local') + balancer.add(n) + settle('healthy') + verify_public() + except Exception as original: + failed = [] + # Restore a healthy old-image node first; it can serve its peer's rollback. + for n in reversed(changed): + try: + peer = next(other for other in nodes if other is not n) + if not peer.status()['ready']: + raise RolloutError('No healthy peer for rollback') + balancer.remove(n) + n.route('peer') + settle('drain') + n.activate(before[n.name]['image']) + n.verify() + n.route('local') + balancer.add(n) + settle('healthy') + except Exception: + failed.append(n.name) + record({'status': 'rollback_failed' if failed else 'rolled_back', + 'image': image, 'failed_nodes': failed}) + raise RolloutError('Deployment failed; rollback failed on ' + ','.join(failed) + if failed else 'Deployment failed; previous images restored') from original + record({'status': 'deployed', 'image': image}) diff --git a/workspace/operations/digitalocean/cd/tests/test_guards.py b/workspace/operations/digitalocean/cd/tests/test_guards.py new file mode 100644 index 000000000..f7a102d55 --- /dev/null +++ b/workspace/operations/digitalocean/cd/tests/test_guards.py @@ -0,0 +1,47 @@ +import sys +from pathlib import Path +from types import SimpleNamespace +import unittest +from unittest.mock import patch +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from migration_plan import plan +import deploy + + +class MigrationTests(unittest.TestCase): + def script(self, safe): + return SimpleNamespace(get_heads=lambda: ['084'], + iterate_revisions=lambda head,current: [SimpleNamespace(revision='084', + module=SimpleNamespace(deployment_backward_compatible=safe))]) + def test_unmarked_migration_blocks(self): + with self.assertRaises(RuntimeError): plan(self.script(False), ['083']) + def test_safe_migration_is_identified(self): + self.assertEqual(plan(self.script(True), ['083'])['pending'], ['084']) + def test_multiple_current_revisions_block(self): + with self.assertRaises(RuntimeError): plan(self.script(True), ['082','083']) + def test_current_schema_needs_no_backup(self): + s=SimpleNamespace(get_heads=lambda:['083'],iterate_revisions=lambda *_:[]) + self.assertEqual(plan(s,['083'])['pending'], []) + + +class CommandGuards(unittest.TestCase): + def test_ci_cannot_activate_or_change_routing(self): + for args in [['activate','openagents-backend:release-'+'a'*40], ['route','peer'], ['sh','-c','id'], ['run','a'*40]]: + with self.assertRaises(ValueError): deploy.dispatch('ci',args) + def test_peer_cannot_start_deployment_or_stage_image(self): + for verb in ['deploy','stage','result']: + with self.assertRaises(ValueError): deploy.dispatch('peer',[verb,'a'*40]) + def test_commit_injection_and_path_traversal_rejected(self): + for sha in ['../secret','a'*40+';id','$(id)','a'*39,'A'*40]: + with self.assertRaises(ValueError): deploy.stage(sha) + def test_image_injection_rejected_before_docker(self): + with patch.object(deploy,'command') as cmd: + for image in ['ubuntu:latest','openagents-backend:release-'+ 'a'*40+';id','../secret']: + with self.assertRaises(ValueError): deploy.validate(image) + cmd.assert_not_called() + def test_image_revision_must_match(self): + with patch.object(deploy,'command',return_value='b'*40): + with self.assertRaises(ValueError): deploy.validate('openagents-backend:release-'+'a'*40) + + +if __name__ == '__main__': unittest.main() diff --git a/workspace/operations/digitalocean/cd/tests/test_rollout.py b/workspace/operations/digitalocean/cd/tests/test_rollout.py new file mode 100644 index 000000000..fce43e275 --- /dev/null +++ b/workspace/operations/digitalocean/cd/tests/test_rollout.py @@ -0,0 +1,92 @@ +import sys +from pathlib import Path +import unittest +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from rollout import rollout, RolloutError + + +class Node: + def __init__(self, name, calls): + self.name, self.calls, self.image = name, calls, 'old' + self.ok = True + self.fail_activate = False + self.fail_verify = False + self.fail_rollback = False + def status(self): return {'image': self.image, 'ready': self.ok} + def validate(self, image): self.calls.append((self.name, 'validate', image)) + def route(self, mode): self.calls.append((self.name, 'route', mode)) + def activate(self, image): + self.calls.append((self.name, 'activate', image)); self.image = image + if (image == 'new' and self.fail_activate) or (image == 'old' and self.fail_rollback): + self.ok = False; raise RuntimeError('bad candidate') + self.ok = True + def verify(self): + if self.image == 'new' and self.fail_verify: raise RuntimeError('smoke failed') + + +class Balancer: + def __init__(self, nodes, calls): self.members = set(n.name for n in nodes); self.calls = calls + def contains_all(self, nodes): return self.members == {n.name for n in nodes} + def remove(self, n): + self.calls.append((n.name, 'remove')); self.members.discard(n.name) + assert self.members, 'last backend removed' + def add(self, n): self.calls.append((n.name, 'add')); self.members.add(n.name) + + +class RolloutTests(unittest.TestCase): + def setUp(self): + self.calls, self.records = [], [] + self.nodes = [Node('secondary', self.calls), Node('primary', self.calls)] + self.lb = Balancer(self.nodes, self.calls) + self.migrations = 0 + def migrate(self): self.migrations += 1; self.calls.append(('migrate',)) + def run_rollout(self, migrate=None, verify=None): + return rollout(self.nodes, self.lb, 'new', migrate or self.migrate, + verify or (lambda: None), self.records.append, lambda phase: None) + def test_secondary_first_and_migration_once(self): + self.run_rollout() + self.assertEqual(self.migrations, 1) + self.assertEqual([x[0] for x in self.calls if len(x)>1 and x[1]=='activate'], ['secondary','primary']) + self.assertLess(self.calls.index(('migrate',)), self.calls.index(('secondary','remove'))) + self.assertEqual(self.records[-1]['status'], 'deployed') + def test_unhealthy_peer_blocks_any_change(self): + self.nodes[1].ok = False + with self.assertRaises(RolloutError): self.run_rollout() + self.assertEqual(self.migrations, 0) + self.assertFalse(any('remove' in x for x in self.calls)) + def test_migration_failure_keeps_both_nodes_untouched(self): + def fail(): raise RuntimeError('migration failed') + with self.assertRaises(RuntimeError): self.run_rollout(migrate=fail) + self.assertEqual([n.image for n in self.nodes], ['old','old']) + self.assertFalse(any('remove' in x for x in self.calls)) + def test_failed_secondary_restores_it_and_never_replaces_primary(self): + self.nodes[0].fail_activate=True + with self.assertRaises(RolloutError): self.run_rollout() + self.assertEqual([n.image for n in self.nodes], ['old','old']) + self.assertNotIn(('primary','activate','new'), self.calls) + self.assertEqual(self.records[-1]['status'], 'rolled_back') + def test_failed_primary_restores_broken_node_before_secondary(self): + self.nodes[1].fail_activate=True + with self.assertRaises(RolloutError): self.run_rollout() + self.assertEqual([n.image for n in self.nodes], ['old','old']) + self.assertEqual(self.records[-1]['status'], 'rolled_back') + def test_smoke_failure_rolls_back(self): + self.nodes[0].fail_verify=True + with self.assertRaises(RolloutError): self.run_rollout() + self.assertEqual([n.image for n in self.nodes], ['old','old']) + def test_public_failure_rolls_back_both(self): + def fail(): raise RuntimeError('public failed') + with self.assertRaises(RolloutError): self.run_rollout(verify=fail) + self.assertEqual([n.image for n in self.nodes], ['old','old']) + def test_failed_rollback_stays_out_of_pool(self): + self.nodes[0].fail_activate=True;self.nodes[0].fail_rollback=True + with self.assertRaises(RolloutError): self.run_rollout() + self.assertEqual(self.lb.members, {'primary'}) + self.assertEqual(self.records[-1]['status'], 'rollback_failed') + def test_unexpected_pool_membership_blocks_deployment(self): + self.lb.members.add('unexpected') + with self.assertRaises(RolloutError): self.run_rollout() + self.assertEqual(self.migrations, 0) + + +if __name__ == '__main__': unittest.main()