setup.py

#!/usr/bin/env python3
"""Ghosted setup wizard: provision AWS infra and write config.toml."""
from __future__ import annotations
import sys
from pathlib import Path

from ghosted import awsbuild, provision

CONFIG_TEMPLATE = """\
site_title = "{site_title}"
site_description = "{site_description}"
author = "{author}"
prompt = "{prompt}"
footer_tagline = "{footer_tagline}"
base_url = "{base_url}"
posts_per_page = 25
rss_count = 50

[aws]
region = "{region}"
site_bucket = "{site_bucket}"
images_bucket = "{images_bucket}"
cloudfront_distribution_id = "{cloudfront_distribution_id}"
cloudfront_domain = "{cloudfront_domain}"
oidc_role_arn = "{oidc_role_arn}"
profile = "{profile}"

[github]
owner = "{owner}"
repo = "{repo}"
"""


def write_config(values: dict, path="config.toml") -> None:
    Path(path).write_text(CONFIG_TEMPLATE.format(**values), encoding="utf-8")


def configure_domain(domain, *, acm, route53, cf, dist_id, dist_domain,
                     attach_only=False, sleep=None):
    apex = awsbuild.is_apex(domain)
    aliases = awsbuild.domain_aliases(domain)
    sans = ["www." + domain] if apex else []
    cert_arn = provision.ensure_certificate(acm, domain, sans)
    zone_id = provision.find_hosted_zone(route53, domain)
    cf_url = f"https://{dist_domain}"
    wait_kw = {"sleep": sleep} if sleep else {}

    if zone_id and not attach_only:
        records = provision.cert_validation_records(acm, cert_arn)
        if records:
            provision.upsert_records(route53, zone_id,
                                     awsbuild.validation_cname_changes(records))
        if not provision.wait_certificate(acm, cert_arn, **wait_kw):
            print("Certificate not validated yet. Re-run: python setup.py --attach-domain")
            return {"base_url": cf_url, "status": "pending"}
    elif not zone_id and not attach_only:
        print(f"\nYour DNS is not in Route 53. Add these validation records at your DNS host:")
        for r in provision.cert_validation_records(acm, cert_arn):
            print(f"  {r['Name']}  {r['Type']}  ->  {r['Value']}")
        print("Then run: python setup.py --attach-domain")
        return {"base_url": cf_url, "status": "manual_pending"}
    elif attach_only:
        if provision.describe_status(acm, cert_arn) != "ISSUED":
            print("Certificate still not ISSUED — add the validation records and wait.")
            return {"base_url": cf_url, "status": "manual_pending"}

    # attach cert + aliases, regenerate the (domain-aware) function
    provision.attach_domain_to_distribution(cf, dist_id, aliases, cert_arn)
    code = awsbuild.cf_function_code(domain if apex else None).encode()
    provision.ensure_function(cf, "ghosted-rewrite", code)

    if zone_id:
        provision.upsert_records(route53, zone_id,
                                 awsbuild.route53_alias_changes(aliases, dist_domain))
    else:
        print("\nPoint your domain at CloudFront by adding at your DNS host:")
        for a in aliases:
            print(f"  {a}  ->  {dist_domain}   (ALIAS, or CNAME for a subdomain)")
    return {"base_url": f"https://{domain}", "status": "done"}


def run(answers: dict, *, s3, cf, iam, account_id: str, plan: bool = False,
        acm=None, route53=None) -> dict:
    site_b, img_b = answers["site_bucket"], answers["images_bucket"]
    region = answers["region"]
    owner, repo = answers["owner"], answers["repo"]
    if plan:
        for line in (
            f"create/verify S3 bucket {site_b} (private, {region})",
            f"create/verify S3 bucket {img_b} (private, {region})",
            "create/verify CloudFront OAC, function, distribution",
            f"create/verify OIDC provider + role ghosted-deploy for {owner}/{repo}",
            "write config.toml",
        ):
            print("PLAN:", line)
        return {}

    provision.ensure_bucket(s3, site_b, region)
    provision.ensure_bucket(s3, img_b, region)
    oac = provision.ensure_oac(cf, "ghosted-oac")
    apex = awsbuild.is_apex(answers["domain"]) if answers.get("domain") else False
    code = awsbuild.cf_function_code(answers["domain"] if (answers.get("domain") and apex) else None).encode()
    fn_arn = provision.ensure_function(cf, "ghosted-rewrite", code)
    dist_cfg = awsbuild.distribution_config(
        site_bucket=site_b, images_bucket=img_b, region=region, oac_id=oac,
        function_arn=fn_arn, caller_reference=f"ghosted-{site_b}")
    dist_id, dist_domain, dist_arn = provision.ensure_distribution(cf, dist_cfg)
    provision.put_bucket_policy(s3, site_b, awsbuild.bucket_policy(site_b, dist_arn))
    provision.put_bucket_policy(s3, img_b, awsbuild.bucket_policy(img_b, dist_arn))
    provision.ensure_oidc_provider(iam, account_id)
    role_arn = provision.ensure_deploy_role(
        iam, "ghosted-deploy",
        awsbuild.oidc_trust_policy(account_id, owner, repo),
        awsbuild.deploy_role_policy(account_id, site_b, img_b, dist_id))

    base_url = f"https://{dist_domain}"
    if answers.get("domain"):
        dom = configure_domain(answers["domain"], acm=acm, route53=route53, cf=cf,
                               dist_id=dist_id, dist_domain=dist_domain)
        base_url = dom["base_url"]

    return {
        "site_title": answers["site_title"], "site_description": answers["site_description"],
        "author": answers["author"], "prompt": answers["prompt"],
        "footer_tagline": answers["footer_tagline"],
        "base_url": base_url,
        "region": region, "site_bucket": site_b, "images_bucket": img_b,
        "cloudfront_distribution_id": dist_id, "cloudfront_domain": dist_domain,
        "oidc_role_arn": role_arn,
        "profile": answers.get("profile", ""), "owner": owner, "repo": repo,
    }


def _ask(label: str, default: str = "") -> str:
    suffix = f" [{default}]" if default else ""
    val = input(f"{label}{suffix}: ").strip()
    return val or default


def main() -> int:
    import boto3
    plan = "--plan" in sys.argv
    attach_only = "--attach-domain" in sys.argv

    # parse --domain <value>
    cli_domain = None
    if "--domain" in sys.argv:
        idx = sys.argv.index("--domain")
        if idx + 1 < len(sys.argv):
            cli_domain = sys.argv[idx + 1]

    try:
        ident = boto3.client("sts").get_caller_identity()
    except Exception as e:  # noqa: BLE001
        print("ERROR: no working AWS credentials. Run `aws configure` first.\n"
              "See the README for creating an IAM user + access key.\n", e, file=sys.stderr)
        return 1
    account_id = ident["Account"]
    print(f"AWS account: {account_id}\n")

    if attach_only:
        import tomllib
        cfg = tomllib.loads(Path("config.toml").read_text(encoding="utf-8"))
        dist_id = cfg["aws"]["cloudfront_distribution_id"]
        dist_domain = cfg["aws"].get("cloudfront_domain") or cfg["base_url"].removeprefix("https://")
        domain = cli_domain or cfg.get("domain") or _ask("Domain to attach")
        acm = boto3.client("acm", region_name="us-east-1")
        route53 = boto3.client("route53")
        cf = boto3.client("cloudfront")
        res = configure_domain(domain, acm=acm, route53=route53, cf=cf,
                               dist_id=dist_id, dist_domain=dist_domain, attach_only=True)
        if res.get("status") == "done":
            c = tomllib.loads(Path("config.toml").read_text(encoding="utf-8"))
            vals = {
                "site_title": c["site_title"], "site_description": c["site_description"],
                "author": c["author"], "prompt": c["prompt"],
                "footer_tagline": c["footer_tagline"], "base_url": res["base_url"],
                "region": c["aws"]["region"], "site_bucket": c["aws"]["site_bucket"],
                "images_bucket": c["aws"]["images_bucket"],
                "cloudfront_distribution_id": c["aws"]["cloudfront_distribution_id"],
                "cloudfront_domain": c["aws"].get("cloudfront_domain", ""),
                "oidc_role_arn": c["aws"]["oidc_role_arn"],
                "profile": c["aws"].get("profile", ""),
                "owner": c["github"]["owner"], "repo": c["github"]["repo"],
            }
            write_config(vals)
        print(res.get("status"))
        return 0

    answers = {
        "site_title": _ask("Site title", "My Ghosted Blog"),
        "site_description": _ask("Site description", "A static blog you own."),
        "author": _ask("Author", "Your Name"),
        "prompt": _ask("Header prompt", "$ cat blog"),
        "footer_tagline": _ask("Footer tagline", "powered by Ghosted"),
        "region": _ask("AWS region", "us-east-1"),
        "site_bucket": _ask("Site bucket name"),
        "images_bucket": _ask("Images bucket name"),
        "owner": _ask("GitHub owner"),
        "repo": _ask("GitHub repo"),
        "profile": "",
    }
    if cli_domain:
        answers["domain"] = cli_domain
    elif not plan:
        d = _ask("Custom domain (blank to skip)", "")
        if d:
            answers["domain"] = d

    region = answers["region"]
    s3 = boto3.client("s3", region_name=region)
    cf = boto3.client("cloudfront")
    iam = boto3.client("iam")
    acm = boto3.client("acm", region_name="us-east-1")
    route53 = boto3.client("route53")
    values = run(answers, s3=s3, cf=cf, iam=iam, account_id=account_id, plan=plan,
                 acm=acm, route53=route53)
    if plan:
        print("\n(plan only — nothing created)")
        return 0
    write_config(values)
    print(f"\n✅ Provisioned. Site will be live at {values['base_url']} once you push.\n"
          "Next: git add config.toml && git commit -m 'configure' && git push")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())