Alicloud Compute Swas Open

interface:
  display_name: "Alibaba Cloud Compute SWAS Open"
  short_description: "Simple app server operations workflows"
  default_prompt: "Use $alicloud-compute-swas-open to complete this compute/swas task on Alibaba Cloud."

#!/usr/bin/env python3
"""Fix SSH access for SWAS instances.

- Ensures authorized_keys contains the provided public key
- Ensures PermitRootLogin and PubkeyAuthentication are enabled
- Restarts ssh service
- Optionally sets Port in sshd_config

This script uses SWAS RunCommand to execute on the instance.
"""

from __future__ import annotations

import argparse
import os
from alibabacloud_swas_open20200601.client import Client as SwasClient
from alibabacloud_swas_open20200601 import models as swas_models
from alibabacloud_tea_openapi import models as open_api_models


def create_client(region_id: str) -> SwasClient:
    config = open_api_models.Config(
        region_id=region_id,
        endpoint=f"swas.{region_id}.aliyuncs.com",
    )
    ak = os.getenv("ALICLOUD_ACCESS_KEY_ID") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
    sk = os.getenv("ALICLOUD_ACCESS_KEY_SECRET") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
    token = os.getenv("ALICLOUD_SECURITY_TOKEN") or os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN")
    if ak and sk:
        config.access_key_id = ak
        config.access_key_secret = sk
        if token:
            config.security_token = token
    return SwasClient(config)


def build_script(pub_key: str, user: str, port: str | None) -> str:
    port_block = ""
    if port:
        port_block = f"""
if ! grep -q '^Port' $SSHD_CONFIG; then
  echo 'Port {port}' >> $SSHD_CONFIG
else
  sed -i 's/^Port.*/Port {port}/' $SSHD_CONFIG
fi
"""

    return f"""#!/bin/bash
set -e
USER_NAME="{user}"
HOME_DIR=$(getent passwd "$USER_NAME" | cut -d: -f6)
if [ -z "$HOME_DIR" ]; then
  echo "User $USER_NAME not found"
  exit 1
fi

mkdir -p "$HOME_DIR/.ssh"
chmod 700 "$HOME_DIR/.ssh"
if ! grep -qF '{pub_key}' "$HOME_DIR/.ssh/authorized_keys" 2>/dev/null; then
  echo '{pub_key}' >> "$HOME_DIR/.ssh/authorized_keys"
fi
chmod 600 "$HOME_DIR/.ssh/authorized_keys"
chown -R "$USER_NAME":"$USER_NAME" "$HOME_DIR/.ssh"

SSHD_CONFIG=/etc/ssh/sshd_config
if ! grep -q '^PermitRootLogin' $SSHD_CONFIG; then
  echo 'PermitRootLogin yes' >> $SSHD_CONFIG
else
  sed -i 's/^PermitRootLogin.*/PermitRootLogin yes/' $SSHD_CONFIG
fi
if ! grep -q '^PubkeyAuthentication' $SSHD_CONFIG; then
  echo 'PubkeyAuthentication yes' >> $SSHD_CONFIG
else
  sed -i 's/^PubkeyAuthentication.*/PubkeyAuthentication yes/' $SSHD_CONFIG
fi
{port_block}

if systemctl list-unit-files | grep -q '^ssh\\.service'; then
  systemctl restart ssh
elif systemctl list-unit-files | grep -q '^sshd\\.service'; then
  systemctl restart sshd
elif command -v service >/dev/null 2>&1; then
  service ssh restart || service sshd restart
elif [ -x /etc/init.d/ssh ]; then
  /etc/init.d/ssh restart
elif [ -x /etc/init.d/sshd ]; then
  /etc/init.d/sshd restart
else
  echo "No ssh service found"
  exit 1
fi
"""


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--region", required=True)
    parser.add_argument("--instance-id", required=True)
    parser.add_argument("--user", default="root")
    parser.add_argument("--port", help="Set SSH port in sshd_config")
    parser.add_argument("--pubkey", default="~/.ssh/id_ed25519.pub")
    args = parser.parse_args()

    pubkey_path = os.path.expanduser(args.pubkey)
    with open(pubkey_path, "r", encoding="utf-8") as f:
        pub_key = f.read().strip()

    script = build_script(pub_key, args.user, args.port)
    client = create_client(args.region)
    resp = client.run_command(swas_models.RunCommandRequest(
        region_id=args.region,
        instance_id=args.instance_id,
        name="fix-ssh-access",
        type="RunShellScript",
        command_content=script,
    ))
    print("InvokeId:", resp.body.invoke_id)
    return 0


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

#!/usr/bin/env python3
"""Get SSH port from sshd_config via SWAS RunCommand."""

from __future__ import annotations

import argparse
import os
from alibabacloud_swas_open20200601.client import Client as SwasClient
from alibabacloud_swas_open20200601 import models as swas_models
from alibabacloud_tea_openapi import models as open_api_models


def create_client(region_id: str) -> SwasClient:
    config = open_api_models.Config(
        region_id=region_id,
        endpoint=f"swas.{region_id}.aliyuncs.com",
    )
    ak = os.getenv("ALICLOUD_ACCESS_KEY_ID") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
    sk = os.getenv("ALICLOUD_ACCESS_KEY_SECRET") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
    token = os.getenv("ALICLOUD_SECURITY_TOKEN") or os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN")
    if ak and sk:
        config.access_key_id = ak
        config.access_key_secret = sk
        if token:
            config.security_token = token
    return SwasClient(config)


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--region", required=True)
    parser.add_argument("--instance-id", required=True)
    args = parser.parse_args()

    script = """#!/bin/bash
set -e
PORT=$(grep -E '^Port ' /etc/ssh/sshd_config | tail -n 1 | awk '{print $2}')
if [ -z "$PORT" ]; then
  PORT=22
fi
echo $PORT
"""

    client = create_client(args.region)
    resp = client.run_command(swas_models.RunCommandRequest(
        region_id=args.region,
        instance_id=args.instance_id,
        name="get-ssh-port",
        type="RunShellScript",
        command_content=script,
    ))
    print("InvokeId:", resp.body.invoke_id)
    return 0


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

#!/usr/bin/env python3
"""List SWAS instances across all regions.

Outputs TSV by default. Use --json for JSON output.
"""

from __future__ import annotations

import argparse
import json
import os
import sys

from alibabacloud_swas_open20200601.client import Client as SwasClient
from alibabacloud_swas_open20200601 import models as swas_models
from alibabacloud_tea_openapi import models as open_api_models


def create_client(region_id: str) -> SwasClient:
    config = open_api_models.Config(
        region_id=region_id,
        endpoint=f"swas.{region_id}.aliyuncs.com",
    )
    ak = os.getenv("ALICLOUD_ACCESS_KEY_ID") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
    sk = os.getenv("ALICLOUD_ACCESS_KEY_SECRET") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
    token = os.getenv("ALICLOUD_SECURITY_TOKEN") or os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN")
    if ak and sk:
        config.access_key_id = ak
        config.access_key_secret = sk
        if token:
            config.security_token = token
    return SwasClient(config)


def list_regions() -> list[str]:
    client = create_client("cn-hangzhou")
    resp = client.list_regions(swas_models.ListRegionsRequest())
    return [r.region_id for r in resp.body.regions]


def list_instances(region_id: str):
    client = create_client(region_id)
    resp = client.list_instances(swas_models.ListInstancesRequest(region_id=region_id))
    return resp.body.instances or []


def to_record(region_id: str, inst) -> dict:
    return {
        "region_id": region_id,
        "instance_id": inst.instance_id,
        "instance_name": getattr(inst, "instance_name", None),
        "status": getattr(inst, "status", None),
        "public_ip": getattr(inst, "public_ip_address", None),
        "inner_ip": getattr(inst, "inner_ip_address", None),
        "plan_id": getattr(inst, "plan_id", None),
        "plan_name": getattr(inst, "plan_name", None),
        "cpu": getattr(inst, "cpu", None),
        "memory_gib": getattr(inst, "memory", None),
        "zone_id": getattr(inst, "zone_id", None),
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--json", action="store_true", help="Output JSON array")
    parser.add_argument("--output", help="Write output to file")
    args = parser.parse_args()

    records = []
    failed_regions: list[tuple[str, str]] = []
    for region_id in list_regions():
        try:
            instances = list_instances(region_id)
        except Exception as exc:
            failed_regions.append((region_id, str(exc)))
            print(f"Warning: failed to query region {region_id}: {exc}", file=sys.stderr)
            continue
        for inst in instances:
            records.append(to_record(region_id, inst))

    if args.json:
        output = json.dumps(records, ensure_ascii=False, indent=2)
    else:
        lines = [
            "region_id\tinstance_id\tinstance_name\tstatus\tpublic_ip\tinner_ip\tplan_name\tplan_id\tcpu\tmemory_gib\tzone_id"
        ]
        for r in records:
            lines.append(
                "\t".join(
                    [
                        str(r.get("region_id") or ""),
                        str(r.get("instance_id") or ""),
                        str(r.get("instance_name") or ""),
                        str(r.get("status") or ""),
                        str(r.get("public_ip") or ""),
                        str(r.get("inner_ip") or ""),
                        str(r.get("plan_name") or ""),
                        str(r.get("plan_id") or ""),
                        str(r.get("cpu") or ""),
                        str(r.get("memory_gib") or ""),
                        str(r.get("zone_id") or ""),
                    ]
                )
            )
        output = "\n".join(lines)

    if args.output:
        with open(args.output, "w", encoding="utf-8") as f:
            f.write(output)
    else:
        print(output)

    if failed_regions:
        print("\nFailed regions:", file=sys.stderr)
        for region_id, err in failed_regions:
            print(f"- {region_id}: {err}", file=sys.stderr)

    return 0


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

#!/usr/bin/env python3
"""Summarize SWAS instances by plan across all regions."""

from __future__ import annotations

import argparse
import os
from collections import Counter

from alibabacloud_swas_open20200601.client import Client as SwasClient
from alibabacloud_swas_open20200601 import models as swas_models
from alibabacloud_tea_openapi import models as open_api_models


def create_client(region_id: str) -> SwasClient:
    config = open_api_models.Config(
        region_id=region_id,
        endpoint=f"swas.{region_id}.aliyuncs.com",
    )
    ak = os.getenv("ALICLOUD_ACCESS_KEY_ID") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
    sk = os.getenv("ALICLOUD_ACCESS_KEY_SECRET") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
    token = os.getenv("ALICLOUD_SECURITY_TOKEN") or os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN")
    if ak and sk:
        config.access_key_id = ak
        config.access_key_secret = sk
        if token:
            config.security_token = token
    return SwasClient(config)


def list_regions() -> list[str]:
    client = create_client("cn-hangzhou")
    resp = client.list_regions(swas_models.ListRegionsRequest())
    return [r.region_id for r in resp.body.regions]


def list_instances(region_id: str):
    client = create_client(region_id)
    resp = client.list_instances(swas_models.ListInstancesRequest(region_id=region_id))
    return resp.body.instances or []


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--output", help="Write output to file")
    args = parser.parse_args()

    counter: Counter[str] = Counter()
    for region_id in list_regions():
        for inst in list_instances(region_id):
            plan = getattr(inst, "plan_name", None) or getattr(inst, "plan_id", None) or "(unknown)"
            counter[plan] += 1

    lines = ["plan\tcount"]
    for plan, cnt in sorted(counter.items()):
        lines.append(f"{plan}\t{cnt}")

    output = "\n".join(lines) if len(lines) > 1 else "No SWAS instances found."
    if args.output:
        with open(args.output, "w", encoding="utf-8") as f:
            f.write(output)
    else:
        print(output)

    return 0


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

#!/usr/bin/env python3
"""Summarize SWAS instances by status across all regions."""

from __future__ import annotations

import argparse
import os
from collections import Counter

from alibabacloud_swas_open20200601.client import Client as SwasClient
from alibabacloud_swas_open20200601 import models as swas_models
from alibabacloud_tea_openapi import models as open_api_models


def create_client(region_id: str) -> SwasClient:
    config = open_api_models.Config(
        region_id=region_id,
        endpoint=f"swas.{region_id}.aliyuncs.com",
    )
    ak = os.getenv("ALICLOUD_ACCESS_KEY_ID") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
    sk = os.getenv("ALICLOUD_ACCESS_KEY_SECRET") or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
    token = os.getenv("ALICLOUD_SECURITY_TOKEN") or os.getenv("ALIBABA_CLOUD_SECURITY_TOKEN")
    if ak and sk:
        config.access_key_id = ak
        config.access_key_secret = sk
        if token:
            config.security_token = token
    return SwasClient(config)


def list_regions() -> list[str]:
    client = create_client("cn-hangzhou")
    resp = client.list_regions(swas_models.ListRegionsRequest())
    return [r.region_id for r in resp.body.regions]


def list_instances(region_id: str):
    client = create_client(region_id)
    resp = client.list_instances(swas_models.ListInstancesRequest(region_id=region_id))
    return resp.body.instances or []


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--output", help="Write output to file")
    args = parser.parse_args()

    counter: Counter[str] = Counter()
    for region_id in list_regions():
        for inst in list_instances(region_id):
            status = getattr(inst, "status", None) or "(unknown)"
            counter[status] += 1

    lines = ["status\tcount"]
    for status, cnt in sorted(counter.items()):
        lines.append(f"{status}\t{cnt}")

    output = "\n".join(lines) if len(lines) > 1 else "No SWAS instances found."
    if args.output:
        with open(args.output, "w", encoding="utf-8") as f:
            f.write(output)
    else:
        print(output)

    return 0


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