# core
## 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
``` python
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
``` python
def Multipass(
args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):
```
*Wrap multipass CLI: manage local Ubuntu VMs for testing*
``` python
Multipass().vms()
```
[]
## Cloud-init generation
[`vps_init()`](https://vedicreader.github.io/vpseasy/core.html#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()`](https://vedicreader.github.io/vpseasy/core.html#multi_init)
is the local equivalent via `cloud_init_base` — same Docker setup, no
UFW, no fail2ban.
------------------------------------------------------------------------
source
### vps_init
``` python
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
``` python
def gen_key(
slug, key_dir:NoneType=None
):
```
*Generate ed25519 key pair at /. Overwrites existing.
Returns AttrDict(key, pub, pub_str).*
------------------------------------------------------------------------
source
### keygen
``` python
def keygen(
path
):
```
*Call self as a function.*
------------------------------------------------------------------------
source
### multi_init
``` python
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).*
``` python
# 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
``` python
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`](https://vedicreader.github.io/vpseasy/core.html#hetzner)
wraps the hcloud Python SDK — no CLI binary or config files required.
Token read from `HCLOUD_TOKEN` by default.
------------------------------------------------------------------------
source
### Hetzner
``` python
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.
``` python
hz = Hetzner()
L(hz.servers()).attrgot('name')
```
['vedicreader-cx32-hel', 'lego']
``` python
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()`](https://vedicreader.github.io/vpseasy/core.html#deploy)
syncs a Compose stack to a remote host and brings it up.
------------------------------------------------------------------------
source
### deploy
``` python
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
``` python
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
``` python
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
``` python
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.*
``` python
# _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()`](https://vedicreader.github.io/vpseasy/core.html#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
``` python
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
``` python
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.*
``` python
# 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
``` python
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
``` python
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()`](https://vedicreader.github.io/vpseasy/core.html#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()`](https://vedicreader.github.io/vpseasy/core.html#vps_init)
and
[`multi_init()`](https://vedicreader.github.io/vpseasy/core.html#multi_init).
------------------------------------------------------------------------
source
### load_pub_keys
``` python
def load_pub_keys(
paths:NoneType=None
)->list:
```
*Load SSH public key strings. paths=None → auto-detect from
~/.ssh/id\_*.pub\*
``` python
import tempfile
```
``` python
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`](https://vedicreader.github.io/vpseasy/core.html#multi_init),
then exercises
[`wait_ssh`](https://vedicreader.github.io/vpseasy/core.html#wait_ssh),
[`chk_cloud_init`](https://vedicreader.github.io/vpseasy/core.html#chk_cloud_init),
[`chk_docker`](https://vedicreader.github.io/vpseasy/core.html#chk_docker),
[`run_ssh`](https://vedicreader.github.io/vpseasy/core.html#run_ssh),
and [`sync`](https://vedicreader.github.io/vpseasy/core.html#sync).
``` python
# --- 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
``` python
# 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
``` python
import tempfile
```
``` python
# 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
``` python
# 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
``` python
# --- teardown ---
mp.rm(_VM); print('VM removed')
```
VM removed
## Docker Compose helpers
[`vols_to_binds()`](https://vedicreader.github.io/vpseasy/core.html#vols_to_binds)
converts absolute container paths to local bind mounts.
[`caddy_stack()`](https://vedicreader.github.io/vpseasy/core.html#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
``` python
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
``` python
def vols_to_binds(
vols
):
```
*Convert \[“/app/data”\] → \[“./data:/app/data”\] for docker compose
bind mounts*
``` python
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
``` python
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
``` python
def mv_skill_md(
dry_run:bool=True, dir:NoneType=None
)->None:
```
*Copy bundled SKILL.md to .agents/skills/vpseasy/ and
~/.claude/skills/vpseasy/*
``` python
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