core

VPS provisioning: cloud-init generation, Hetzner hcloud CLI wrapper, SSH deployment helpers

Multipass for local testing

Using multipass to test cloud-init and deployment locally before provisioning real VPSes. You’ll need to install Multipass and have it in your PATH for this to work.

---

source

deploy_mp

def deploy_mp(
    name, src, path:str='/srv/app', key:NoneType=None, build:bool=True, verbose:bool=True
)->None:

Sync local directory into a Multipass VM and run docker compose up -d.

---

source

Multipass

def Multipass(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Wrap multipass CLI: manage local Ubuntu VMs for testing

Multipass().vms()

[]

Cloud-init generation

vps_init() builds a cloud-init YAML for a fresh VPS via fastcloudinit.cloud_init_config: UFW hardening, user creation, SSH key setup, optional Docker. multi_init() is the local equivalent via cloud_init_base — same Docker setup, no UFW, no fail2ban.

---

source

vps_init

def vps_init(
    hostname, pub_keys:NoneType=None, username:str='deploy', docker:bool=True, pkgs:NoneType=None,
    cmds:NoneType=None, kw:VAR_KEYWORD
):

Cloud-init for a fresh VPS. pub_keys=None → auto-generates ed25519 key pair. Returns AttrDict(yaml, key).

---

source

gen_key

def gen_key(
    slug, key_dir:NoneType=None
):

Generate ed25519 key pair at /. Overwrites existing. Returns AttrDict(key, pub, pub_str).

---

source

keygen

def keygen(
    path
):

Call self as a function.

---

source

multi_init

def multi_init(
    hostname, pub_keys:NoneType=None, username:str='deploy', docker:bool=True, pkgs:NoneType=None,
    cmds:NoneType=None
):

Cloud-init for Multipass local VMs. pub_keys=None → auto-generates ed25519 key pair. Returns AttrDict(yaml, key).

# explicit pub_keys → no key generated, key=None
ci = multi_init('mylocal', 'ssh-rsa AAAA...')
assert '#cloud-config' in ci.yaml and 'get.docker.com' in ci.yaml and 'ufw' not in ci.yaml
assert 'power_state' in ci.yaml and ci.key is None
print(ci.yaml)

# auto-generate mode: key pair written to ~/.ssh/mylocal{,.pub}
ci2 = multi_init('mylocal', docker=False)
assert 'get.docker.com' not in ci2.yaml and 'power_state' not in ci2.yaml
assert ci2.key is not None and ci2.key.exists()
assert Path(str(ci2.key) + '.pub').exists()
ci2.key.unlink(); Path(str(ci2.key) + '.pub').unlink()
print('multi_init OK')

#cloud-config
hostname: mylocal
preserve_hostname: false
packages:
- curl
package_update: true
package_upgrade: true
disable_root: true
ssh_pwauth: false
users:
- name: deploy
  groups:
  - sudo
  shell: /bin/bash
  sudo:
  - ALL=(ALL) NOPASSWD:ALL
  ssh_authorized_keys:
  - ssh-rsa AAAA...
runcmd:
- curl -fsSL https://get.docker.com | sh
- usermod -aG docker deploy
- systemctl enable --now docker
power_state:
  mode: reboot
  message: Rebooting
  timeout: 1
  condition: true

multi_init OK

ci = vps_init('myserver', 'ssh-rsa AAAA...', docker=True)
assert '#cloud-config' in ci.yaml and 'get.docker.com' in ci.yaml
assert 'fail2ban' in ci.yaml and 'unattended-upgrades' in ci.yaml
assert ci.key is None  # explicit pub_keys → no key generated
print(ci.yaml)
print('vps_init OK')

#cloud-config
hostname: myserver
preserve_hostname: false
packages:
- curl
- fail2ban
- unattended-upgrades
package_update: true
package_upgrade: true
disable_root: true
ssh_pwauth: false
users:
- name: deploy
  groups:
  - sudo
  shell: /bin/bash
  sudo:
  - ALL=(ALL) NOPASSWD:ALL
  ssh_authorized_keys:
  - ssh-rsa AAAA...
runcmd:
- curl -fsSL https://get.docker.com | sh
- usermod -aG docker deploy
- systemctl enable --now docker
- ufw default deny incoming
- ufw default allow outgoing
- ufw logging off
- ufw allow 22/tcp
- ufw --force enable
apt:
  conf: 'APT::Periodic::Update-Package-Lists "1";

    APT::Periodic::Download-Upgradeable-Packages "1";

    APT::Periodic::AutocleanInterval "7";

    APT::Periodic::Unattended-Upgrade "0";

    Unattended-Upgrade::Automatic-Reboot "false";

    '
write_files:
- path: /etc/logrotate.d/00-cloud-init-global
  owner: root:root
  permissions: '0644'
  content: "/var/log/*.log {\n    weekly\n    rotate 7\n    compress\n    su root adm\n    create\n    missingok\n}\n"
power_state:
  mode: reboot
  message: Rebooting
  timeout: 1
  condition: true

vps_init OK

Hetzner (hcloud Python SDK)

Hetzner wraps the hcloud Python SDK — no CLI binary or config files required. Token read from HCLOUD_TOKEN by default.

---

source

Hetzner

def Hetzner(
    token:NoneType=None
):

Hetzner Cloud VPS provider via hcloud Python SDK. Token read from HCLOUD_TOKEN by default.

Integration tests

Requires HCLOUD_TOKEN in the environment. Creates a minimal cx11 server, verifies the lifecycle, then deletes it.

hz = Hetzner()
L(hz.servers()).attrgot('name')

['vedicreader-cx32-hel', 'lego']

sn = 'fastops-test'
hz.delete(sn)
svr=hz.create(sn, server_type='cx23', location='hel1')
assert svr.ip, 'create() should return an IP'
print(f'create OK: {svr.ip}')
o = lambda: L(hz.servers()).attrgot('name')
print(f'servers() OK: {o()}')
assert sn in o(), f'servers() should include {sn}'
assert hz.server_ip(sn) == svr.ip
hz.delete(sn)
assert sn not in o()
print(f'servers() after delete: {o()}')
print('delete() OK\nAll hcloud tests passed!')

create OK: 95.216.194.42
servers() OK: ['vedicreader-cx32-hel', 'vedicreader']
servers() after delete: ['vedicreader-cx32-hel']
delete() OK
All hcloud tests passed!

SSH helpers

Pure subprocess-based SSH/rsync utilities — no paramiko dependency. deploy() syncs a Compose stack to a remote host and brings it up.

---

source

deploy

def deploy(
    host, src:str='.', path:str='/srv/app', user:str='deploy', build:bool=True, key:NoneType=None,
    name:NoneType=None, include:NoneType=None, exclude:NoneType=None, extra:NoneType=None, verbose:bool=False
):

Sync src to host via rsync then docker compose up if docker is available. extra= extra rsync flags forwarded to sync().

---

source

chk_docker

def chk_docker(
    host, u:str='deploy', k:NoneType=None, name:NoneType=None, verbose:bool=False
)->bool:

Verify docker daemon is running and user can run containers.

---

source

sync

def sync(
    host, src:str='.', path:str='/srv/app', user:str='deploy', key:NoneType=None, name:NoneType=None,
    include:NoneType=None, exclude:NoneType=None, extra:NoneType=None, verbose:bool=False
):

Rsync local src to remote host:path. include= whitelist patterns, exclude= blacklist patterns. extra= extra rsync flags e.g. “–checksum” or [“–ignore-times”,“–partial”].

---

source

run_ssh

def run_ssh(
    host, cmds:VAR_POSITIONAL, user:str='deploy', key:NoneType=None, name:NoneType=None, port:int=22,
    check:bool=True, verbose:bool=False, same_in_win:bool=False, ignore_ex:bool=False, as_bytes:bool=False,
    stderr:bool=True, kwargs:VAR_KEYWORD
):

Run commands on remote host via SSH. Returns stdout string. check=False ignores non-zero exit codes.

# _ssh_base: pure flag construction
assert _ssh('1.2.3.4', 'deploy', None, 22) == ['ssh', '-o', 'StrictHostKeyChecking=accept-new', 'deploy@1.2.3.4']
assert _ssh('1.2.3.4', 'deploy', '/k', 2222)[-3:] == ['-p', '2222', 'deploy@1.2.3.4']
assert '-i' in _ssh('h', 'u', '/my/key', 22)
print('_ssh_base OK')
assert _res_key('/tmp/mykey') == '/tmp/mykey'
try: _res_key(name='__no_such_vpseasy_key__')
except FileNotFoundError as e: assert '__no_such_vpseasy_key__' in str(e)
print('_res_key OK')

_ssh_base OK
_res_key OK

Verification helpers

Cloud-init runs asynchronously after create() returns. Use wait_ssh() to block until the server is reachable, then check_cloud_init() to confirm the bootstrap completed successfully before deploying.

---

source

chk_cloud_init

def chk_cloud_init(
    host, u:str='deploy', k:NoneType=None, name:NoneType=None, verbose:bool=True
)->str:

Return cloud-init status: done|running|error|unknown. check=False handles exit code 2 (done-with-warnings) on Ubuntu 24.04.

---

source

wait_ssh

def wait_ssh(
    host, u:str='deploy', k:NoneType=None, name:NoneType=None, p:int=22, tout:int=300, interval:int=5,
    verbose:bool=True
):

Poll SSH until connection succeeds or raises TimeoutError.

# chk_docker: returns False when key lookup fails (FileNotFoundError caught internally — no network call)
assert chk_docker('localhost', name='__no_such_vpseasy_key__') is False
print('chk_docker: bad key → False OK')

# wait_ssh: raises TimeoutError immediately when tout=0
try: wait_ssh('1.2.3.4', tout=0); assert False
except TimeoutError as e: assert '1.2.3.4' in str(e)
print('wait_ssh: tout=0 → TimeoutError OK')

chk_docker: bad key → False OK
wait_ssh: tout=0 → TimeoutError OK

---

source

wait_ready

def wait_ready(
    host, u:str='deploy', k:NoneType=None, name:NoneType=None, tout:int=300, interval:int=5, retries:int=2,
    verbose:bool=False
):

Wait for SSH then poll cloud-init until done. Retries cloud-init polling up to retries times before raising TimeoutError.

---

source

hetzner_deploy

def hetzner_deploy(
    name, src, hz:NoneType=None, image:str='ubuntu-24.04', server_type:str='cx23', location:NoneType=None,
    path:str='/srv/app', build:bool=True, include:NoneType=None, exclude:NoneType=None, extra:NoneType=None,
    tout:int=600, retries:int=2, verbose:bool=True
):

Full pipeline: provision Hetzner VPS (idempotent) → wait for cloud-init → deploy. Returns AttrDict(ip, name, key).

SSH key helpers

load_pub_keys() resolves a list of paths (or auto-detects ~/.ssh/id_*.pub) into a flat list of public key strings ready for vps_init() and multi_init().

---

source

load_pub_keys

def load_pub_keys(
    paths:NoneType=None
)->list:

Load SSH public key strings. paths=None → auto-detect from ~/.ssh/id_.pub*

import tempfile

keys = load_pub_keys()
print(f'Found {len(keys)} local SSH key(s)')

_f = Path(tempfile.mktemp(suffix='.pub'))
_f.write_text('ssh-ed25519 TESTKEY comment')
assert load_pub_keys([str(_f)]) == ['ssh-ed25519 TESTKEY comment']
_f.unlink()
assert load_pub_keys(['/no_such_key.pub']) == []
print('load_pub_keys OK')

# gen_key: creates ed25519 pair, returns AttrDict, overwrites existing
_d = Path(tempfile.mkdtemp())
kp = gen_key('testkey', key_dir=_d)
assert kp.key.exists() and kp.pub.exists()
assert kp.pub_str[0].startswith('ssh-ed25519')
kp2 = gen_key('testkey', key_dir=_d)  # overwrite — should not raise
assert kp2.pub_str[0].startswith('ssh-ed25519')
import shutil; shutil.rmtree(_d)
print('gen_key OK')

Found 0 local SSH key(s)
load_pub_keys OK
gen_key OK

Integration tests: SSH helpers and verification helpers

Requires Multipass. Launches a minimal Ubuntu VM with local SSH keys injected via multi_init, then exercises wait_ssh, chk_cloud_init, chk_docker, run_ssh, and sync.

# --- setup: auto-generate key pair, inject via multi_init, launch VM ---
_VM = 'testvm'
mp = Multipass()
try: mp.rm(_VM)
except: pass

ci = multi_init(_VM, docker=False) # generates ~/.ssh/fastops-vpstest{,.pub}
vm = mp.launch(_VM, image='24.04', cpus=1, memory='512M', disk='5G', cloud_init=ci)
ip = mp.ip(vm.name)
_key = vm.key
print(f'VM at {ip}, key: {_key}')

Creating testvm  Configuring testvm  Starting testvm  Waiting for initialization to complete  Launched: testvm
VM at 192.168.2.62, key: /Users/71293/.ssh/testvm

# wait_ssh / chk_cloud_init / run_ssh
assert wait_ssh(ip, k=_key, tout=30) is True; print('wait_ssh OK')
status = chk_cloud_init(ip, name=_VM, verbose=True)
assert status in ('done', 'running'), f'unexpected: {status!r}'; print(f'chk_cloud_init: {status}')
assert run_ssh(ip, 'echo hi', key=_key).strip() == 'hi'; print('run_ssh OK')

SSH to host 192.168.2.62 check succeeded
wait_ssh OK
Resolved SSH key from name slug: /Users/71293/.ssh/testvm
Ran SSH command on 192.168.2.62: sudo cloud-init status → (0, 'status: done')
chk_cloud_init: done
run_ssh OK

import tempfile

# sync: full dir / exclude / include (whitelist)
_src = Path(tempfile.mkdtemp())
(_src/'a.txt').write_text('hello')
(_src/'b.log').write_text('log')
(_src/'sub').mkdir(); (_src/'sub'/'c.txt').write_text('sub')

deploy_mp(_VM, _src, path='/tmp/app')  # sanity check deploy_mp with build=False
assert run_ssh(ip, 'ls /tmp/app', key=_key).split() == ['a.txt', 'b.log', 'sub']

sync(ip, _src, '/tmp/t1', key=_key)
assert run_ssh(ip, 'cat /tmp/t1/a.txt', key=_key).strip() == 'hello'; print('sync full OK')

sync(ip, _src, '/tmp/t2', key=_key, exclude=['*.log'])
assert run_ssh(ip, 'ls /tmp/t2', key=_key).split() == ['a.txt', 'sub']; print('sync exclude OK')

sync(ip, _src, '/tmp/t3', key=_key, include=['a.txt'])
assert run_ssh(ip, 'ls /tmp/t3', key=_key).strip() == 'a.txt'; print('sync include OK')

sync(ip, _src, '/tmp/t4', key=_key, include=['sub/'])
assert run_ssh(ip, 'cat /tmp/t4/sub/c.txt', key=_key).strip() == 'sub'; print('sync include-dir OK')

Resolved SSH key from name slug: /Users/71293/.ssh/testvm
Ensured remote path /tmp/app exists and is writable by deploy
Resolved SSH key from name slug: /Users/71293/.ssh/testvm
Running rsync: rsync -az --delete -e ssh -o StrictHostKeyChecking=accept-new -i /Users/71293/.ssh/testvm /var/folders/kg/9vdw4mdd1fs58svgh4k1qhr09x7dqh/T/tmpjs007i2_/ deploy@192.168.2.62:/tmp/app/
Rsync completed successfully
Resolved SSH key from name slug: /Users/71293/.ssh/testvm
Docker check failed: ;; bash: line 1: docker: command not found
sync full OK
sync exclude OK
sync include OK
sync include-dir OK

# extra= forwards arbitrary rsync flags. str and list both accepted via listify().
sync(ip, _src, '/tmp/t5', key=_key, extra='--checksum')
assert run_ssh(ip, 'cat /tmp/t5/a.txt', key=_key).strip() == 'hello'; print('sync extra=str (--checksum, hash-match) OK')

sync(ip, _src, '/tmp/t6', key=_key, extra=['--ignore-times', '--partial'])
assert run_ssh(ip, 'cat /tmp/t6/a.txt', key=_key).strip() == 'hello'; print('sync extra=list (--ignore-times re-copies all) OK')

# extra= composes with include/exclude
sync(ip, _src, '/tmp/t7', key=_key, extra='--checksum', exclude=['*.log'])
assert run_ssh(ip, 'ls /tmp/t7', key=_key).split() == ['a.txt', 'sub']; print('sync extra + exclude OK')

sync extra=str (--checksum, hash-match) OK
sync extra=list (--ignore-times re-copies all) OK
sync extra + exclude OK

# --- teardown ---
mp.rm(_VM); print('VM removed')

VM removed

Docker Compose helpers

vols_to_binds() converts absolute container paths to local bind mounts. caddy_stack() generates a full production Compose file: app + caddy + optional cloudflared, web network, and caddy volumes — one call replaces the boilerplate in every FastHTML deploy.

---

source

caddy_stack

def caddy_stack(
    domain, df, vols:NoneType=None, env_file:str='.env', cloudflared:bool=True, root:NoneType=None,
    conf:NoneType=None, kw:VAR_KEYWORD
):

Compose with app + caddy + optional cloudflared, web network, caddy volumes. df: Dockerfile instance. root: if given, saves Dockerfile + docker-compose.yml there. **kw passed to caddy_svc.

---

source

vols_to_binds

def vols_to_binds(
    vols
):

Convert [“/app/data”] → [“./data:/app/data”] for docker compose bind mounts

assert vols_to_binds(['/app/data', '/app/backups']) == ['./data:/app/data', './backups:/app/backups']
assert vols_to_binds('/app/data') == ['./data:/app/data']
print('vols_to_binds OK')

_d = Path(tempfile.mkdtemp())
c = caddy_stack('myapp.example.com', fasthtml_app(), vols=['/app/data'], root=_d, conf=str(_d/'Caddyfile'))
d = c.to_dict()
assert set(d['services']) == {'app', 'caddy', 'cloudflared'}
assert 'web' in d['networks']
assert 'caddy_data' in d['volumes'] and 'caddy_config' in d['volumes']
assert d['services']['app']['volumes'] == ['./data:/app/data']
assert (_d/'Dockerfile').exists() and (_d/'docker-compose.yml').exists() and (_d/'Caddyfile').exists()
print('caddy_stack OK')

vols_to_binds OK
caddy_stack OK

caddy_stack('myapp.example.com', fasthtml_app(), vols=['/app/data'])

services:
  app:
    build: .
    volumes:
    - ./data:/app/data
    env_file:
    - .env
    restart: unless-stopped
    networks:
    - web
  caddy:
    image: caddy:2
    depends_on:
    - app
    volumes:
    - ./Caddyfile:/etc/caddy/Caddyfile
    - caddy_data:/data
    - caddy_config:/config
    networks:
    - web
    restart: unless-stopped
  cloudflared:
    image: cloudflare/cloudflared:latest
    command: tunnel --no-autoupdate run --url http://caddy
    environment:
    - TUNNEL_TOKEN=${CF_TUNNEL_TOKEN}
    networks:
    - web
    restart: unless-stopped
networks:
  web: null
volumes:
  caddy_data: null
  caddy_config: null

---

source

mv_skill_md

def mv_skill_md(
    dry_run:bool=True, dir:NoneType=None
)->None:

Copy bundled SKILL.md to .agents/skills/vpseasy/ and ~/.claude/skills/vpseasy/

import io, sys, tempfile, os
_d = Path(tempfile.mkdtemp())
(_d/'SKILL.md').write_text('test-skill')   # provide a src so function doesn't bail early
_saved_cwd = os.getcwd(); os.chdir(_d)
_buf = io.StringIO(); sys.stdout, _saved = _buf, sys.stdout
mv_skill_md(dry_run=True, dir=str(_d))
sys.stdout = _saved; os.chdir(_saved_cwd)
_out = _buf.getvalue()
assert '.agents/skills/vpseasy/SKILL.md' in _out, _out
assert '.claude/skills/vpseasy/SKILL.md' in _out, _out
print('mv_skill_md dry_run OK')

mv_skill_md dry_run OK