432 lines
16 KiB
Python
Executable File
432 lines
16 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""Read-only by default. Reuse/prepare the fixed Beijing host via the aliyun CLI."""
|
||
|
||
import argparse
|
||
import base64
|
||
import fcntl
|
||
import hashlib
|
||
import json
|
||
import math
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import uuid
|
||
from pathlib import Path
|
||
|
||
REGION = "cn-beijing"
|
||
PUBLIC_IP = "123.56.71.98"
|
||
|
||
|
||
class CloudError(RuntimeError):
|
||
pass
|
||
|
||
|
||
class AliyunCLI:
|
||
def __init__(self, profile=None):
|
||
self.binary = shutil.which("aliyun")
|
||
if not self.binary:
|
||
raise CloudError(
|
||
"aliyun CLI is not installed; configure CLI credentials locally, never paste secrets into chat"
|
||
)
|
||
self.profile = profile
|
||
|
||
def __call__(self, product, action, **params):
|
||
cmd = [self.binary, product, action, "--RegionId", REGION]
|
||
if self.profile:
|
||
cmd += ["--profile", self.profile]
|
||
for key, value in params.items():
|
||
cmd += [
|
||
"--" + key,
|
||
json.dumps(value, separators=(",", ":"))
|
||
if isinstance(value, (list, dict, bool))
|
||
else str(value),
|
||
]
|
||
try:
|
||
result = subprocess.run(
|
||
cmd, capture_output=True, text=True, timeout=60, check=True
|
||
)
|
||
data = json.loads(result.stdout)
|
||
if not isinstance(data, dict):
|
||
raise CloudError(
|
||
f"{product} {action} returned an invalid response shape"
|
||
)
|
||
return data
|
||
except (
|
||
subprocess.CalledProcessError,
|
||
subprocess.TimeoutExpired,
|
||
json.JSONDecodeError,
|
||
) as exc:
|
||
# CLI diagnostics can contain request details; don't log raw stdout/stderr or credentials.
|
||
raise CloudError(
|
||
f"{product} {action} failed; inspect the authenticated CLI locally (no mutation retry here)"
|
||
) from exc
|
||
|
||
|
||
def identity(cfg):
|
||
if cfg.get("region") != REGION or cfg.get("public_ip") != PUBLIC_IP:
|
||
raise CloudError("region/IP must remain cn-beijing / 123.56.71.98")
|
||
if (
|
||
not isinstance(cfg.get("project_tag"), str)
|
||
or not cfg["project_tag"]
|
||
or len(cfg["project_tag"]) > 64
|
||
):
|
||
raise CloudError("a dedicated project_tag is required")
|
||
|
||
|
||
def parse_count(value):
|
||
try:
|
||
count = int(value)
|
||
except (TypeError, ValueError) as exc:
|
||
raise CloudError("invalid inventory count; no mutations allowed") from exc
|
||
if isinstance(value, bool) or count < 0:
|
||
raise CloudError("invalid inventory count; no mutations allowed")
|
||
return count
|
||
|
||
|
||
def instances(api, **filters):
|
||
found = []
|
||
seen = set()
|
||
for page in range(1, 21):
|
||
data = api("ecs", "DescribeInstances", PageSize=100, PageNumber=page, **filters)
|
||
rows = data.get("Instances", {}).get("Instance")
|
||
if not isinstance(rows, list):
|
||
raise CloudError("malformed instance inventory; no mutations allowed")
|
||
for row in rows:
|
||
key = row.get("InstanceId") if isinstance(row, dict) else None
|
||
if not isinstance(key, str) or not key or key in seen:
|
||
raise CloudError(
|
||
"invalid or changing instance inventory; no mutations allowed"
|
||
)
|
||
seen.add(key)
|
||
found.extend(rows)
|
||
total = data.get("TotalCount")
|
||
if total is not None:
|
||
expected = parse_count(total)
|
||
if len(found) == expected:
|
||
return found
|
||
if len(found) > expected or not rows:
|
||
raise CloudError(
|
||
"incomplete or changing instance inventory; no mutations allowed"
|
||
)
|
||
elif len(rows) < 100:
|
||
return found
|
||
raise CloudError(
|
||
"instance inventory exceeded the safety bound; no mutations allowed"
|
||
)
|
||
|
||
|
||
def eip(api):
|
||
data = api(
|
||
"vpc", "DescribeEipAddresses", EipAddress=PUBLIC_IP, PageSize=100, PageNumber=1
|
||
)
|
||
rows = data.get("EipAddresses", {}).get("EipAddress")
|
||
if not isinstance(rows, list):
|
||
raise CloudError("malformed EIP inventory")
|
||
if parse_count(data.get("TotalCount", len(rows))) != len(rows):
|
||
raise CloudError("incomplete EIP inventory")
|
||
matches = [row for row in rows if row.get("IpAddress") == PUBLIC_IP]
|
||
if len(matches) > 1:
|
||
raise CloudError("ambiguous EIP ownership")
|
||
if matches and not matches[0].get("AllocationId"):
|
||
raise CloudError("EIP allocation ID missing; no creation allowed")
|
||
return matches[0] if matches else None
|
||
|
||
|
||
def owned(row, cfg):
|
||
tags = row.get("Tags", {}).get("Tag", [])
|
||
return row.get("InstanceId") == cfg.get("adopt_instance_id") or any(
|
||
t.get("TagKey") == "project" and t.get("TagValue") == cfg["project_tag"]
|
||
for t in tags
|
||
)
|
||
|
||
|
||
def plan(cfg, api):
|
||
identity(cfg)
|
||
address = eip(api)
|
||
if address and address.get("InstanceId"):
|
||
if address.get("InstanceType") != "EcsInstance":
|
||
raise CloudError(
|
||
"fixed EIP is attached to a non-ECS resource; refusing to rebind"
|
||
)
|
||
rows = instances(api, InstanceIds=[address["InstanceId"]])
|
||
elif address:
|
||
if address.get("Status") != "Available":
|
||
raise CloudError(
|
||
"EIP is not available; wait/reconcile instead of creating a host"
|
||
)
|
||
rows = instances(
|
||
api, **{"Tag.1.Key": "project", "Tag.1.Value": cfg["project_tag"]}
|
||
)
|
||
else:
|
||
rows = instances(api, PublicIpAddresses=[PUBLIC_IP])
|
||
rows = [
|
||
r
|
||
for r in rows
|
||
if PUBLIC_IP in r.get("PublicIpAddress", {}).get("IpAddress", [])
|
||
]
|
||
if not rows:
|
||
raise CloudError(
|
||
"fixed IP is neither an owned EIP nor an existing ECS public IP; cannot recreate this address"
|
||
)
|
||
if len(rows) > 1:
|
||
raise CloudError(
|
||
"multiple candidate instances; explicit reconciliation required"
|
||
)
|
||
if address and address.get("InstanceId") and not rows:
|
||
raise CloudError(
|
||
"EIP attachment has no visible instance; do not create or detach"
|
||
)
|
||
row = rows[0] if rows else None
|
||
if row and not owned(row, cfg):
|
||
raise CloudError(
|
||
"existing instance is not project-owned; set adopt_instance_id only after owner approval"
|
||
)
|
||
return {
|
||
"region": REGION,
|
||
"public_ip": PUBLIC_IP,
|
||
"action": "reuse"
|
||
if row and (not address or address.get("InstanceId"))
|
||
else ("bind" if row else "create_and_bind"),
|
||
"instance_id": row.get("InstanceId") if row else None,
|
||
"instance_status": row.get("Status") if row else None,
|
||
"instance_charge_type": row.get("InstanceChargeType") if row else None,
|
||
"spot_strategy": row.get("SpotStrategy") if row else None,
|
||
"allocation_id": address.get("AllocationId") if address else None,
|
||
"ip_kind": "eip" if address else "instance_public_ip",
|
||
"warning": "spot interruption can terminate calls; fixed instance public IP is not recoverable like an EIP",
|
||
}
|
||
|
||
|
||
def create_params(cfg):
|
||
required = (
|
||
"image_id",
|
||
"instance_type",
|
||
"vswitch_id",
|
||
"security_group_id",
|
||
"key_pair_name",
|
||
)
|
||
if any(
|
||
not isinstance(cfg.get(k), str)
|
||
or not cfg[k].strip()
|
||
or cfg[k].startswith(("CHANGE_ME", "--"))
|
||
for k in required
|
||
):
|
||
raise CloudError(
|
||
"creation requires approved image/type/VSwitch/security group/SSH KeyPair"
|
||
)
|
||
spot_strategy = cfg.get("spot_strategy", "SpotWithPriceLimit")
|
||
if spot_strategy not in ("SpotWithPriceLimit", "SpotAsPriceGo"):
|
||
raise CloudError("spot_strategy must be SpotWithPriceLimit or SpotAsPriceGo")
|
||
price = cfg.get("spot_price_limit")
|
||
if spot_strategy == "SpotWithPriceLimit" and (
|
||
isinstance(price, bool)
|
||
or not isinstance(price, (int, float))
|
||
or not math.isfinite(price)
|
||
or price <= 0
|
||
):
|
||
raise CloudError(
|
||
"a positive, finite spot_price_limit is required for SpotWithPriceLimit"
|
||
)
|
||
disk = cfg.get("system_disk_gib", 40)
|
||
if isinstance(disk, bool) or not isinstance(disk, int) or not 40 <= disk <= 200:
|
||
raise CloudError(
|
||
"system_disk_gib must be within the approved 40–200 GiB safety bound"
|
||
)
|
||
params = {
|
||
"ImageId": cfg["image_id"],
|
||
"InstanceType": cfg["instance_type"],
|
||
"VSwitchId": cfg["vswitch_id"],
|
||
"SecurityGroupId": cfg["security_group_id"],
|
||
"KeyPairName": cfg["key_pair_name"],
|
||
"Amount": 1,
|
||
"InstanceName": cfg["project_tag"],
|
||
"InstanceChargeType": "PostPaid",
|
||
"InternetMaxBandwidthOut": 0,
|
||
"SpotStrategy": spot_strategy,
|
||
"SystemDisk.Category": cfg.get("system_disk_category", "cloud_essd"),
|
||
"SystemDisk.Size": disk,
|
||
"SystemDisk.PerformanceLevel": cfg.get("system_disk_performance_level", "PL1"),
|
||
"Tag.1.Key": "project",
|
||
"Tag.1.Value": cfg["project_tag"],
|
||
}
|
||
if spot_strategy == "SpotWithPriceLimit":
|
||
params["SpotPriceLimit"] = price
|
||
user_data_file = cfg.get("user_data_file")
|
||
if user_data_file is not None:
|
||
if not isinstance(user_data_file, str) or not user_data_file.strip():
|
||
raise CloudError("user_data_file must be a non-empty local file path")
|
||
path = Path(user_data_file)
|
||
if path.is_symlink() or not path.is_file():
|
||
raise CloudError("user_data_file must be an existing non-symlink file")
|
||
try:
|
||
content = path.read_bytes()
|
||
except OSError as exc:
|
||
raise CloudError("user_data_file cannot be read") from exc
|
||
if len(content) > 16 * 1024:
|
||
raise CloudError("user_data_file exceeds the ECS 16 KiB limit")
|
||
params["UserData"] = base64.b64encode(content).decode("ascii")
|
||
return params
|
||
|
||
|
||
def save_state(path, data):
|
||
temporary = path.with_suffix(".tmp")
|
||
fd = os.open(
|
||
temporary, os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_NOFOLLOW, 0o600
|
||
)
|
||
with os.fdopen(fd, "w") as file:
|
||
json.dump(data, file)
|
||
file.flush()
|
||
os.fsync(file.fileno())
|
||
temporary.replace(path)
|
||
directory_fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
|
||
try:
|
||
os.fsync(directory_fd)
|
||
finally:
|
||
os.close(directory_fd)
|
||
|
||
|
||
def wait_running(api, instance_id, sleep=time.sleep):
|
||
for _ in range(60):
|
||
rows = instances(api, InstanceIds=[instance_id])
|
||
if len(rows) == 1 and rows[0].get("Status") == "Running":
|
||
return
|
||
if rows and rows[0].get("Status") in ("Stopped", "Stopping"):
|
||
raise CloudError(
|
||
f"instance {instance_id} is stopped; refusing to start or replace it automatically"
|
||
)
|
||
sleep(5)
|
||
raise CloudError(
|
||
f"instance {instance_id} is not Running; preserve state and retry/reconcile, do not create manually"
|
||
)
|
||
|
||
|
||
def apply(cfg, api, state_path, sleep=time.sleep):
|
||
state_path = Path(state_path)
|
||
state_path.parent.mkdir(parents=True, exist_ok=True)
|
||
if state_path.is_symlink():
|
||
raise CloudError("state file must not be a symlink")
|
||
lock_fd = os.open(
|
||
str(state_path) + ".lock", os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW, 0o600
|
||
)
|
||
with os.fdopen(lock_fd, "w") as lock:
|
||
try:
|
||
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
except BlockingIOError as exc:
|
||
raise CloudError(
|
||
"another provisioning process owns this state file"
|
||
) from exc
|
||
current = plan(cfg, api) # Re-read immediately before mutations.
|
||
instance_id = current["instance_id"]
|
||
if current["action"] == "reuse":
|
||
if current["instance_status"] != "Running":
|
||
raise CloudError(
|
||
"existing host is not Running; no automatic start/replace"
|
||
)
|
||
return current
|
||
if current["action"] == "create_and_bind":
|
||
params = create_params(cfg)
|
||
fingerprint = hashlib.sha256(
|
||
json.dumps([REGION, PUBLIC_IP, params], sort_keys=True).encode()
|
||
).hexdigest()
|
||
try:
|
||
state = (
|
||
json.loads(state_path.read_text()) if state_path.exists() else {}
|
||
)
|
||
except (OSError, ValueError) as exc:
|
||
raise CloudError(
|
||
"state file cannot be read; no creation allowed"
|
||
) from exc
|
||
if not isinstance(state, dict):
|
||
raise CloudError("invalid state; no creation allowed")
|
||
if state and state.get("fingerprint") != fingerprint:
|
||
raise CloudError(
|
||
"creation configuration changed; reconcile old resources/state before proceeding"
|
||
)
|
||
if not state:
|
||
state = {"fingerprint": fingerprint, "client_token": str(uuid.uuid4())}
|
||
save_state(state_path, state) # Persist BEFORE the billable request.
|
||
try:
|
||
uuid.UUID(state.get("client_token", ""))
|
||
except (ValueError, TypeError, AttributeError) as exc:
|
||
raise CloudError(
|
||
"invalid persisted ClientToken; reconcile state before creation"
|
||
) from exc
|
||
result = api(
|
||
"ecs", "RunInstances", ClientToken=state["client_token"], **params
|
||
)
|
||
ids = result.get("InstanceIdSets", {}).get("InstanceIdSet", [])
|
||
if len(ids) != 1:
|
||
raise CloudError(
|
||
"unexpected creation result; preserve ClientToken and reconcile"
|
||
)
|
||
instance_id = ids[0]
|
||
state["instance_id"] = instance_id
|
||
save_state(state_path, state)
|
||
wait_running(api, instance_id, sleep)
|
||
latest = eip(api)
|
||
if not latest or latest.get("AllocationId") != current["allocation_id"]:
|
||
raise CloudError(
|
||
"fixed EIP changed/disappeared; created host preserved, no substitute IP"
|
||
)
|
||
if latest.get("InstanceId") and latest["InstanceId"] != instance_id:
|
||
raise CloudError(
|
||
"EIP became attached to another instance; refusing to detach it"
|
||
)
|
||
if not latest.get("InstanceId"):
|
||
if latest.get("Status") != "Available":
|
||
raise CloudError(
|
||
"EIP is not safely bindable; preserve instance and retry later"
|
||
)
|
||
api(
|
||
"vpc",
|
||
"AssociateEipAddress",
|
||
AllocationId=latest["AllocationId"],
|
||
InstanceId=instance_id,
|
||
InstanceType="EcsInstance",
|
||
)
|
||
for _ in range(30):
|
||
bound = eip(api)
|
||
if (
|
||
bound
|
||
and bound.get("InstanceId") == instance_id
|
||
and bound.get("Status") == "InUse"
|
||
):
|
||
return dict(
|
||
current,
|
||
action="ready",
|
||
instance_id=instance_id,
|
||
instance_status="Running",
|
||
)
|
||
sleep(2)
|
||
raise CloudError(
|
||
"EIP binding not confirmed; preserve host/state and reconcile, do not allocate another address"
|
||
)
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("--config", required=True)
|
||
parser.add_argument("--state", default=".local/aliyun-host.json")
|
||
parser.add_argument(
|
||
"--apply",
|
||
action="store_true",
|
||
help="owner-approved billable creation/EIP association",
|
||
)
|
||
args = parser.parse_args()
|
||
try:
|
||
cfg = json.loads(Path(args.config).read_text())
|
||
identity(cfg)
|
||
api = AliyunCLI(cfg.get("profile"))
|
||
result = apply(cfg, api, args.state) if args.apply else plan(cfg, api)
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
except (CloudError, OSError, ValueError) as exc:
|
||
print(f"ERROR: {exc}", file=sys.stderr)
|
||
return 1
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|