diff --git a/CLAUDE.md b/CLAUDE.md index f1b18db..65944c6 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -35,10 +35,50 @@ for technical deep-dives, USECASES.md for expected behaviors. ## Build and test -No build step. No tests. Verify changes by: -1. Reading scripts and checking idempotency -2. Walking through each use case in USECASES.md -3. Cross-checking docs against scripts +No build step. Run e2e tests with: + + uvx pytest tests/test_app_e2e.py -v -x + +### Prerequisites + +- `uv` installed (`curl -LsSf https://astral.sh/uv/install.sh | sh`) +- CLI tools: `bws`, `jq`, `dig`, `ssh`, `curl`, `git` +- bws token: set `BWS_ACCESS_TOKEN` env var or save to `~/.config/hantim/bws-token` +- Vultr API key must allow requests from the machine's IP (check https://my.vultr.com/settings/#settingsapi) + +### Test structure + +- `tests/conftest.py` -- Fixtures: preflight checks, bws secrets, API helper, cleanup +- `tests/test_app_e2e.py` -- E2E tests for `tools/app.sh` against real services +- `tests/test_security_headers.py` -- Verify security headers on live sites +- `tests/test_default_server.py` -- Verify unknown Host headers are dropped +- `tests/test_tls.py` -- Verify TLS hardening + +### E2E tests (`test_app_e2e.py`) + +Tests use `kgfamily.com` as a throwaway test domain. Two phases: + +1. **TestAppSubcommands** -- Runs each `app.sh` subcommand individually (dns, repo, + files, cert, garage, build, monitor), verifies side effects via API, then cleans up. + Does not test `verify` (needs full deploy flow). +2. **TestAppAll** -- Runs `app.sh all kgfamily.com`, verifies everything including + site liveness and git commit, then cleans up (including reverting the commit+push). + +Cleanup is done directly via APIs (Vultr, Gitea, Garage, UptimeRobot) + local file +deletion. Does NOT use `remove-app.sh` to avoid coupling. Runs before and after each +phase to handle leftovers from crashed runs. + +### Known issue + +- Vultr API keys can be IP-restricted. If `test_dns` fails with HTTP 401, add the + machine's IP to the Vultr API access control list. + +### Other tests + +The security/TLS/default-server tests run against live sites and are triggered +automatically after nginx deploy. They can also be run locally: + + uvx pytest tests/ -v --ignore=tests/test_app_e2e.py ## Commit style diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..a9b4803 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,265 @@ +"""Shared fixtures and helpers for e2e tests.""" + +import json +import os +import shutil +import subprocess +import urllib.error +import urllib.parse +import urllib.request + +import pytest + +TEST_DOMAIN = "kgfamily.com" +CONTAINER_NAME = TEST_DOMAIN.replace(".", "_") +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) +APP_SH = os.path.join(REPO_ROOT, "tools", "app.sh") + +VULTR_API = "https://api.vultr.com/v2" +GITEA_URL = "https://git.timothykim.net" +GITEA_ORG = "hantim" +GARAGE_API = "https://garage.hantim.net" +UPTIMEROBOT_API = "https://api.uptimerobot.com/v3" + + +def api(method, url, headers=None, data=None): + """HTTP request helper. Returns (status_code, parsed_json_or_None).""" + hdrs = dict(headers or {}) + body = None + if data is not None: + body = json.dumps(data).encode() + hdrs.setdefault("Content-Type", "application/json") + req = urllib.request.Request(url, method=method, headers=hdrs, data=body) + try: + resp = urllib.request.urlopen(req, timeout=30) + raw = resp.read().decode() + return resp.status, json.loads(raw) if raw.strip() else None + except urllib.error.HTTPError as e: + raw = e.read().decode() + try: + return e.code, json.loads(raw) if raw.strip() else None + except json.JSONDecodeError: + return e.code, None + + +def _bws_env(): + """Return env dict with BWS_ACCESS_TOKEN set from token file if needed.""" + env = os.environ.copy() + if not env.get("BWS_ACCESS_TOKEN") and os.path.isfile(BWS_TOKEN_FILE): + env["BWS_ACCESS_TOKEN"] = open(BWS_TOKEN_FILE).read().strip() + return env + + +def run_app(command, domain=TEST_DOMAIN, timeout=300): + """Run app.sh with given command and domain. Returns CompletedProcess.""" + return subprocess.run( + [APP_SH, command, domain], + capture_output=True, text=True, timeout=timeout, + cwd=REPO_ROOT, + ) + + +BWS_TOKEN_FILE = os.path.join( + os.environ.get("XDG_CONFIG_HOME", os.path.expanduser("~/.config")), + "hantim", "bws-token", +) + +REQUIRED_TOOLS = ["bws", "jq", "dig", "ssh", "curl", "git"] + +REQUIRED_SECRETS = [ + "hantim-vultr-api-key", + "hantim-new-app-script", + "hantim-deploy-ssh-private-key", + "hantim-garage-admin-token", + "hantim-garage-media-key-id", + "hantim-uptimerobot-api-key", +] + + +@pytest.fixture(scope="session", autouse=True) +def preflight(): + """Check that all required tools and credentials are available.""" + errors = [] + + for tool in REQUIRED_TOOLS: + if shutil.which(tool) is None: + errors.append(f"missing CLI tool: {tool}") + + if not os.environ.get("BWS_ACCESS_TOKEN") and not os.path.isfile(BWS_TOKEN_FILE): + errors.append( + f"no bws token: set BWS_ACCESS_TOKEN or create {BWS_TOKEN_FILE}" + ) + + if errors: + pytest.exit( + "Preflight failed:\n " + "\n ".join(errors), + returncode=1, + ) + + +@pytest.fixture(scope="session") +def secrets(preflight): + """Fetch all bws secrets once for the session.""" + result = subprocess.run( + ["bws", "secret", "list"], + capture_output=True, text=True, env=_bws_env(), + ) + if result.returncode != 0: + pytest.exit( + f"bws secret list failed (is the token valid?):\n{result.stderr}", + returncode=1, + ) + + all_secrets = json.loads(result.stdout) + by_key = {s["key"]: s["value"] for s in all_secrets} + + missing = [k for k in REQUIRED_SECRETS if k not in by_key] + if missing: + pytest.exit( + f"Missing secrets in Bitwarden: {', '.join(missing)}", + returncode=1, + ) + + return by_key + + +def cleanup(secrets, revert_git=False): + """Best-effort cleanup of all TEST_DOMAIN resources.""" + errors = [] + + # --- UptimeRobot monitor --- + try: + key = secrets["hantim-uptimerobot-api-key"] + auth = {"Authorization": f"Bearer {key}"} + status, data = api( + "GET", + f"{UPTIMEROBOT_API}/monitors?" + + urllib.parse.urlencode({"search": TEST_DOMAIN}), + headers=auth, + ) + if status == 200 and data: + for m in data.get("monitors", []): + if m.get("url") == f"https://www.{TEST_DOMAIN}": + api("DELETE", f"{UPTIMEROBOT_API}/monitors/{m['id']}", headers=auth) + except Exception as e: + errors.append(f"UptimeRobot: {e}") + + # --- Garage bucket --- + try: + token = secrets["hantim-garage-admin-token"] + auth = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} + status, data = api( + "POST", + f"{GARAGE_API}/v2/GetBucketInfo", + headers=auth, + data={"globalAlias": TEST_DOMAIN}, + ) + if status == 200 and data and data.get("id"): + bid = data["id"] + # Deny media key access + media_key = secrets.get("hantim-garage-media-key-id", "") + if media_key: + api( + "POST", + f"{GARAGE_API}/v2/DenyBucketKey", + headers=auth, + data={ + "bucketId": bid, + "accessKeyId": media_key, + "permissions": {"read": True, "write": True, "owner": False}, + }, + ) + # Remove global alias + api( + "POST", + f"{GARAGE_API}/v2/RemoveBucketAlias", + headers=auth, + data={"bucketId": bid, "globalAlias": TEST_DOMAIN}, + ) + # Delete bucket (id is a query param, not body) + api( + "POST", + f"{GARAGE_API}/v2/DeleteBucket?" + + urllib.parse.urlencode({"id": bid}), + headers={"Authorization": f"Bearer {token}"}, + ) + except Exception as e: + errors.append(f"Garage: {e}") + + # --- Gitea repo --- + try: + token = secrets["hantim-new-app-script"] + api( + "DELETE", + f"{GITEA_URL}/api/v1/repos/{GITEA_ORG}/{TEST_DOMAIN}", + headers={"Authorization": f"token {token}"}, + ) + except Exception as e: + errors.append(f"Gitea: {e}") + + # --- Local files --- + local_paths = [ + os.path.join(REPO_ROOT, "docker", TEST_DOMAIN), + os.path.join(REPO_ROOT, ".gitea", "workflows", f"deploy-{TEST_DOMAIN}.yml"), + os.path.join(REPO_ROOT, "docker", "nginx", "conf.d", f"{TEST_DOMAIN}.conf"), + ] + + if revert_git: + # Files may be tracked — check if last commit added them + try: + log = subprocess.run( + ["git", "log", "--format=%s", "-1"], + capture_output=True, text=True, cwd=REPO_ROOT, + ) + if log.stdout.strip() == f"add {TEST_DOMAIN}": + removed_any = False + for f in [ + f"docker/{TEST_DOMAIN}", + f".gitea/workflows/deploy-{TEST_DOMAIN}.yml", + f"docker/nginx/conf.d/{TEST_DOMAIN}.conf", + ]: + full = os.path.join(REPO_ROOT, f) + if os.path.exists(full): + subprocess.run( + ["git", "rm", "-rf", f], + cwd=REPO_ROOT, + capture_output=True, + ) + removed_any = True + if removed_any: + subprocess.run( + ["git", "commit", "-m", f"remove {TEST_DOMAIN}"], + cwd=REPO_ROOT, check=True, capture_output=True, + ) + subprocess.run( + ["git", "push"], + cwd=REPO_ROOT, check=True, capture_output=True, + ) + except Exception as e: + errors.append(f"Git revert: {e}") + + # Always try to remove untracked local files too + for path in local_paths: + try: + if os.path.isdir(path): + shutil.rmtree(path) + elif os.path.isfile(path): + os.remove(path) + except FileNotFoundError: + pass + except Exception as e: + errors.append(f"Local {path}: {e}") + + # --- Vultr DNS zone --- + try: + key = secrets["hantim-vultr-api-key"] + api( + "DELETE", + f"{VULTR_API}/domains/{TEST_DOMAIN}", + headers={"Authorization": f"Bearer {key}"}, + ) + except Exception as e: + errors.append(f"Vultr: {e}") + + if errors: + print(f"Cleanup warnings: {'; '.join(errors)}") diff --git a/tests/test_app_e2e.py b/tests/test_app_e2e.py new file mode 100644 index 0000000..4738eaa --- /dev/null +++ b/tests/test_app_e2e.py @@ -0,0 +1,225 @@ +"""End-to-end tests for tools/app.sh against real services. + +Test domain: kgfamily.com +Run: uvx pytest tests/test_app_e2e.py -v -x +""" + +import os +import ssl +import subprocess +import urllib.error +import urllib.parse +import urllib.request + +import pytest + +from conftest import ( + TEST_DOMAIN, + CONTAINER_NAME, + REPO_ROOT, + VULTR_API, + GITEA_URL, + GITEA_ORG, + GARAGE_API, + UPTIMEROBOT_API, + api, + run_app, + cleanup, +) + + +class TestAppSubcommands: + """Test each app.sh subcommand individually, in order.""" + + @pytest.fixture(autouse=True, scope="class") + def _cleanup(self, secrets): + cleanup(secrets, revert_git=True) + yield + cleanup(secrets, revert_git=False) + + def test_dns(self, secrets): + result = run_app("dns") + assert result.returncode == 0, f"app.sh dns failed:\n{result.stdout}\n{result.stderr}" + + auth = {"Authorization": f"Bearer {secrets['hantim-vultr-api-key']}"} + status, data = api("GET", f"{VULTR_API}/domains/{TEST_DOMAIN}/records", headers=auth) + assert status == 200, f"Vultr API returned {status}" + + records = data["records"] + bare_a = [r for r in records if r["type"] == "A" and r["name"] == ""] + www_a = [r for r in records if r["type"] == "A" and r["name"] == "www"] + + assert len(bare_a) == 1, f"Expected 1 bare A record, got {len(bare_a)}" + assert len(www_a) == 1, f"Expected 1 www A record, got {len(www_a)}" + assert bare_a[0]["data"] == www_a[0]["data"], "Bare and www should point to same IP" + + def test_repo(self, secrets): + result = run_app("repo") + assert result.returncode == 0, f"app.sh repo failed:\n{result.stdout}\n{result.stderr}" + + auth = {"Authorization": f"token {secrets['hantim-new-app-script']}"} + status, data = api("GET", f"{GITEA_URL}/api/v1/repos/{GITEA_ORG}/{TEST_DOMAIN}", headers=auth) + assert status == 200, f"Repo not found (HTTP {status})" + assert data["name"] == TEST_DOMAIN + + def test_files(self): + result = run_app("files") + assert result.returncode == 0, f"app.sh files failed:\n{result.stdout}\n{result.stderr}" + + # compose.yml + compose_path = os.path.join(REPO_ROOT, "docker", TEST_DOMAIN, "compose.yml") + assert os.path.isfile(compose_path) + compose = open(compose_path).read() + assert f"image: git.timothykim.net/hantim/{TEST_DOMAIN}:latest" in compose + assert f"container_name: {CONTAINER_NAME}" in compose + assert "shared" in compose + + # deploy workflow + workflow_path = os.path.join( + REPO_ROOT, ".gitea", "workflows", f"deploy-{TEST_DOMAIN}.yml" + ) + assert os.path.isfile(workflow_path) + workflow = open(workflow_path).read() + assert f"docker/{TEST_DOMAIN}/**" in workflow + assert f"deploy-{TEST_DOMAIN}" in workflow + + # nginx conf + conf_path = os.path.join( + REPO_ROOT, "docker", "nginx", "conf.d", f"{TEST_DOMAIN}.conf" + ) + assert os.path.isfile(conf_path) + conf = open(conf_path).read() + assert f"server_name {TEST_DOMAIN} www.{TEST_DOMAIN}" in conf + assert f"server_name www.{TEST_DOMAIN}" in conf + assert "security-headers.inc" in conf + assert f"proxy_pass http://{CONTAINER_NAME}:80" in conf + + def test_cert(self): + result = run_app("cert", timeout=180) + assert result.returncode == 0, f"app.sh cert failed:\n{result.stdout}\n{result.stderr}" + + def test_garage(self, secrets): + result = run_app("garage") + assert result.returncode == 0, f"app.sh garage failed:\n{result.stdout}\n{result.stderr}" + + auth = { + "Authorization": f"Bearer {secrets['hantim-garage-admin-token']}", + "Content-Type": "application/json", + } + status, data = api( + "POST", + f"{GARAGE_API}/v2/GetBucketInfo", + headers=auth, + data={"globalAlias": TEST_DOMAIN}, + ) + assert status == 200, f"Bucket not found (HTTP {status})" + assert data["id"], "Bucket has no ID" + assert TEST_DOMAIN in data.get("globalAliases", []), "Global alias not set" + website = data.get("websiteAccess", {}) + assert website.get("enabled") is True, f"Website access not enabled: {website}" + + def test_build(self, secrets): + result = run_app("build", timeout=180) + assert result.returncode == 0, f"app.sh build failed:\n{result.stdout}\n{result.stderr}" + + def test_monitor(self, secrets): + result = run_app("monitor") + assert result.returncode == 0, f"app.sh monitor failed:\n{result.stdout}\n{result.stderr}" + + auth = {"Authorization": f"Bearer {secrets['hantim-uptimerobot-api-key']}"} + status, data = api( + "GET", + f"{UPTIMEROBOT_API}/monitors?" + + urllib.parse.urlencode({"search": TEST_DOMAIN}), + headers=auth, + ) + assert status == 200, f"UptimeRobot API returned {status}" + urls = [m["url"] for m in data.get("monitors", [])] + assert f"https://www.{TEST_DOMAIN}" in urls, f"Monitor not found. URLs: {urls}" + + +class TestAppAll: + """Test the full app.sh all flow end-to-end.""" + + @pytest.fixture(autouse=True, scope="class") + def _cleanup(self, secrets): + cleanup(secrets, revert_git=True) + yield + cleanup(secrets, revert_git=True) + + def test_all(self, secrets): + result = run_app("all", timeout=900) + assert result.returncode == 0, f"app.sh all failed:\n{result.stdout}\n{result.stderr}" + + # --- DNS --- + auth = {"Authorization": f"Bearer {secrets['hantim-vultr-api-key']}"} + status, data = api( + "GET", f"{VULTR_API}/domains/{TEST_DOMAIN}/records", headers=auth + ) + assert status == 200 + records = data["records"] + assert any(r["type"] == "A" and r["name"] == "" for r in records) + assert any(r["type"] == "A" and r["name"] == "www" for r in records) + + # --- Gitea repo --- + gitea_auth = {"Authorization": f"token {secrets['hantim-new-app-script']}"} + status, _ = api( + "GET", + f"{GITEA_URL}/api/v1/repos/{GITEA_ORG}/{TEST_DOMAIN}", + headers=gitea_auth, + ) + assert status == 200, "Gitea repo not found" + + # --- Local files --- + assert os.path.isfile( + os.path.join(REPO_ROOT, "docker", TEST_DOMAIN, "compose.yml") + ) + assert os.path.isfile( + os.path.join(REPO_ROOT, ".gitea", "workflows", f"deploy-{TEST_DOMAIN}.yml") + ) + assert os.path.isfile( + os.path.join(REPO_ROOT, "docker", "nginx", "conf.d", f"{TEST_DOMAIN}.conf") + ) + + # --- Git commit was pushed --- + log = subprocess.run( + ["git", "log", "--format=%s", "-1"], + capture_output=True, text=True, cwd=REPO_ROOT, + ) + assert log.stdout.strip() == f"add {TEST_DOMAIN}" + + # --- Garage bucket --- + garage_auth = { + "Authorization": f"Bearer {secrets['hantim-garage-admin-token']}", + "Content-Type": "application/json", + } + status, data = api( + "POST", + f"{GARAGE_API}/v2/GetBucketInfo", + headers=garage_auth, + data={"globalAlias": TEST_DOMAIN}, + ) + assert status == 200, "Garage bucket not found" + + # --- Site is live --- + ctx = ssl.create_default_context() + req = urllib.request.Request(f"https://www.{TEST_DOMAIN}", method="HEAD") + try: + resp = urllib.request.urlopen(req, context=ctx, timeout=10) + assert resp.status in (200, 301, 302) + except urllib.error.HTTPError: + pass # any HTTP response means the site is up + except urllib.error.URLError as e: + pytest.fail(f"Site not reachable: {e}") + + # --- UptimeRobot monitor --- + ur_auth = {"Authorization": f"Bearer {secrets['hantim-uptimerobot-api-key']}"} + status, data = api( + "GET", + f"{UPTIMEROBOT_API}/monitors?" + + urllib.parse.urlencode({"search": TEST_DOMAIN}), + headers=ur_auth, + ) + assert status == 200 + urls = [m["url"] for m in data.get("monitors", [])] + assert f"https://www.{TEST_DOMAIN}" in urls